protected void decodeTags() { current.tagsLength = ByteBufferUtils.readCompressedInt(currentBuffer); if (tagCompressionContext != null) { if (current.uncompressTags) { // Tag compression is been used. uncompress it into tagsBuffer current.ensureSpaceForTags(); try { current.tagsCompressedLength = tagCompressionContext.uncompressTags(currentBuffer, current.tagsBuffer, 0, current.tagsLength); } catch (IOException e) { throw new RuntimeException("Exception while uncompressing tags", e); } } else { ByteBufferUtils.skip(currentBuffer, current.tagsCompressedLength); current.uncompressTags = true;// Reset this. } current.tagsOffset = -1; } else { // When tag compress is not used, let us not do copying of tags bytes into tagsBuffer. // Just mark the tags Offset so as to create the KV buffer later in getKeyValueBuffer() current.tagsOffset = currentBuffer.position(); ByteBufferUtils.skip(currentBuffer, current.tagsLength); } }
@Override public ByteBuffer getFirstKeyInBlock(ByteBuffer block) { block.mark(); block.position(Bytes.SIZEOF_INT); int keyLength = ByteBufferUtils.readCompressedInt(block); ByteBufferUtils.readCompressedInt(block); int commonLength = ByteBufferUtils.readCompressedInt(block); if (commonLength != 0) { throw new AssertionError("Nonzero common length in the first key in " + "block: " + commonLength); } int pos = block.position(); block.reset(); ByteBuffer dup = block.duplicate(); dup.position(pos); dup.limit(pos + keyLength); return dup.slice(); }
@Override public void compressKeyValues(DataOutputStream out, ByteBuffer in, boolean includesMemstoreTS) throws IOException { in.rewind(); ByteBufferUtils.putInt(out, in.limit()); DiffCompressionState previousState = new DiffCompressionState(); DiffCompressionState currentState = new DiffCompressionState(); while (in.hasRemaining()) { compressSingleKeyValue(previousState, currentState, out, in); afterEncodingKeyValue(in, out, includesMemstoreTS); // swap previousState <-> currentState DiffCompressionState tmp = previousState; previousState = currentState; currentState = tmp; } }
@Override public void compressKeyValues(DataOutputStream out, ByteBuffer in, boolean includesMemstoreTS) throws IOException { in.rewind(); ByteBufferUtils.putInt(out, in.limit()); FastDiffCompressionState previousState = new FastDiffCompressionState(); FastDiffCompressionState currentState = new FastDiffCompressionState(); while (in.hasRemaining()) { compressSingleKeyValue(previousState, currentState, out, in); afterEncodingKeyValue(in, out, includesMemstoreTS); // swap previousState <-> currentState FastDiffCompressionState tmp = previousState; previousState = currentState; currentState = tmp; } }
protected void readKeyValueLen() { blockBuffer.mark(); currKeyLen = blockBuffer.getInt(); currValueLen = blockBuffer.getInt(); if (currKeyLen < 0 || currValueLen < 0 || currKeyLen > blockBuffer.limit() || currValueLen > blockBuffer.limit()) { throw new IllegalStateException("Invalid currKeyLen " + currKeyLen + " or currValueLen " + currValueLen + ". Block offset: " + block.getOffset() + ", block length: " + blockBuffer.limit() + ", position: " + blockBuffer.position() + " (without header)."); } ByteBufferUtils.skip(blockBuffer, currKeyLen + currValueLen); if (reader.hfileContext.isIncludesTags()) { // Read short as unsigned, high byte first currTagsLen = ((blockBuffer.get() & 0xff) << 8) ^ (blockBuffer.get() & 0xff); if (currTagsLen < 0 || currTagsLen > blockBuffer.limit()) { throw new IllegalStateException("Invalid currTagsLen " + currTagsLen + ". Block offset: " + block.getOffset() + ", block length: " + blockBuffer.limit() + ", position: " + blockBuffer.position() + " (without header)."); } ByteBufferUtils.skip(blockBuffer, currTagsLen); } readMvccVersion(); blockBuffer.reset(); }
protected void readKeyValueLen() { blockBuffer.mark(); currKeyLen = blockBuffer.getInt(); currValueLen = blockBuffer.getInt(); ByteBufferUtils.skip(blockBuffer, currKeyLen + currValueLen); readMvccVersion(); if (currKeyLen < 0 || currValueLen < 0 || currKeyLen > blockBuffer.limit() || currValueLen > blockBuffer.limit()) { throw new IllegalStateException("Invalid currKeyLen " + currKeyLen + " or currValueLen " + currValueLen + ". Block offset: " + block.getOffset() + ", block length: " + blockBuffer.limit() + ", position: " + blockBuffer.position() + " (without header)."); } blockBuffer.reset(); }
@Override public ByteBuffer getKeyValueBuffer() { ByteBuffer kvBuffer = createKVBuffer(); kvBuffer.putInt(current.keyLength); kvBuffer.putInt(current.valueLength); kvBuffer.put(current.keyBuffer, 0, current.keyLength); ByteBufferUtils.copyFromBufferToBuffer(kvBuffer, currentBuffer, current.valueOffset, current.valueLength); if (current.tagsLength > 0) { // Put short as unsigned kvBuffer.put((byte) (current.tagsLength >> 8 & 0xff)); kvBuffer.put((byte) (current.tagsLength & 0xff)); if (current.tagsOffset != -1) { // the offset of the tags bytes in the underlying buffer is marked. So the temp // buffer,tagsBuffer was not been used. ByteBufferUtils.copyFromBufferToBuffer(kvBuffer, currentBuffer, current.tagsOffset, current.tagsLength); } else { // When tagsOffset is marked as -1, tag compression was present and so the tags were // uncompressed into temp buffer, tagsBuffer. Let us copy it from there kvBuffer.put(current.tagsBuffer, 0, current.tagsLength); } } return kvBuffer; }
protected void readKeyValueLen() { blockBuffer.mark(); currKeyLen = blockBuffer.getInt(); currValueLen = blockBuffer.getInt(); if (currKeyLen < 0 || currValueLen < 0 || currKeyLen > blockBuffer.limit() || currValueLen > blockBuffer.limit()) { throw new IllegalStateException("Invalid currKeyLen " + currKeyLen + " or currValueLen " + currValueLen + ". Block offset: " + block.getOffset() + ", block length: " + blockBuffer.limit() + ", position: " + blockBuffer.position() + " (without header)."); } ByteBufferUtils.skip(blockBuffer, currKeyLen + currValueLen); if (reader.hfileContext.isIncludesTags()) { currTagsLen = blockBuffer.getShort(); if (currTagsLen < 0 || currTagsLen > blockBuffer.limit()) { throw new IllegalStateException("Invalid currTagsLen " + currTagsLen + ". Block offset: " + block.getOffset() + ", block length: " + blockBuffer.limit() + ", position: " + blockBuffer.position() + " (without header)."); } ByteBufferUtils.skip(blockBuffer, currTagsLen); } readMvccVersion(); blockBuffer.reset(); }
@Override public void internalEncodeKeyValues(DataOutputStream out, ByteBuffer in, HFileBlockDefaultEncodingContext encodingCtx) throws IOException { in.rewind(); ByteBufferUtils.putInt(out, in.limit()); DiffCompressionState previousState = new DiffCompressionState(); DiffCompressionState currentState = new DiffCompressionState(); while (in.hasRemaining()) { compressSingleKeyValue(previousState, currentState, out, in); afterEncodingKeyValue(in, out, encodingCtx); // swap previousState <-> currentState DiffCompressionState tmp = previousState; previousState = currentState; currentState = tmp; } }
@Override public void internalEncodeKeyValues(DataOutputStream out, ByteBuffer in, HFileBlockDefaultEncodingContext encodingCtx) throws IOException { in.rewind(); ByteBufferUtils.putInt(out, in.limit()); FastDiffCompressionState previousState = new FastDiffCompressionState(); FastDiffCompressionState currentState = new FastDiffCompressionState(); while (in.hasRemaining()) { compressSingleKeyValue(previousState, currentState, out, in); afterEncodingKeyValue(in, out, encodingCtx); // swap previousState <-> currentState FastDiffCompressionState tmp = previousState; previousState = currentState; currentState = tmp; } }
protected void decodeTags() { current.tagsLength = ByteBufferUtils.readCompressedInt(currentBuffer); if (tagCompressionContext != null) { if (current.uncompressTags) { // Tag compression is been used. uncompress it into tagsBuffer current.ensureSpaceForTags(); try { current.tagsCompressedLength = tagCompressionContext.uncompressTags(currentBuffer, current.tagsBuffer, 0, current.tagsLength); } catch (IOException e) { throw new RuntimeException("Exception while uncompressing tags", e); } } else { ByteBufferUtils.skip(currentBuffer, current.tagsCompressedLength); current.uncompressTags = true;// Reset this. } current.tagsOffset = -1; } else { // When tag compress is not used, let us not do temp copying of tags bytes into tagsBuffer. // Just mark the tags Offset so as to create the KV buffer later in getKeyValueBuffer() current.tagsOffset = currentBuffer.position(); ByteBufferUtils.skip(currentBuffer, current.tagsLength); } }
@Override public ByteBuffer getFirstKeyInBlock(ByteBuffer block) { block.mark(); block.position(Bytes.SIZEOF_INT); int keyLength = ByteBufferUtils.readCompressedInt(block); ByteBufferUtils.readCompressedInt(block); int commonLength = ByteBufferUtils.readCompressedInt(block); if (commonLength != 0) { throw new AssertionError("Nonzero common length in the first key in " + "block: " + commonLength); } int pos = block.position(); block.reset(); return ByteBuffer.wrap(block.array(), block.arrayOffset() + pos, keyLength) .slice(); }
@Test public void testSubBuffer() { ByteBuffer bb1 = ByteBuffer.allocateDirect(10); ByteBuffer bb2 = ByteBuffer.allocateDirect(10); MultiByteBuff multi = new MultiByteBuff(bb1, bb2); long l1 = 1234L, l2 = 100L; multi.putLong(l1); multi.putLong(l2); multi.rewind(); ByteBuffer sub = multi.asSubByteBuffer(Bytes.SIZEOF_LONG); assertEquals(bb1, sub); assertEquals(l1, ByteBufferUtils.toLong(sub, sub.position())); multi.skip(Bytes.SIZEOF_LONG); sub = multi.asSubByteBuffer(Bytes.SIZEOF_LONG); assertNotEquals(bb1, sub); assertNotEquals(bb2, sub); assertEquals(l2, ByteBufferUtils.toLong(sub, sub.position())); multi.rewind(); ObjectIntPair<ByteBuffer> p = new ObjectIntPair<>(); multi.asSubByteBuffer(8, Bytes.SIZEOF_LONG, p); assertNotEquals(bb1, p.getFirst()); assertNotEquals(bb2, p.getFirst()); assertEquals(0, p.getSecond()); assertEquals(l2, ByteBufferUtils.toLong(sub, p.getSecond())); }
public static boolean matchingQualifier(final Cell left, final Cell right) { int lqlength = left.getQualifierLength(); int rqlength = right.getQualifierLength(); if (left instanceof ByteBufferExtendedCell && right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getQualifierByteBuffer(), ((ByteBufferExtendedCell) left).getQualifierPosition(), lqlength, ((ByteBufferExtendedCell) right).getQualifierByteBuffer(), ((ByteBufferExtendedCell) right).getQualifierPosition(), rqlength); } if (left instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getQualifierByteBuffer(), ((ByteBufferExtendedCell) left).getQualifierPosition(), lqlength, right.getQualifierArray(), right.getQualifierOffset(), rqlength); } if (right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) right).getQualifierByteBuffer(), ((ByteBufferExtendedCell) right).getQualifierPosition(), rqlength, left.getQualifierArray(), left.getQualifierOffset(), lqlength); } return Bytes.equals(left.getQualifierArray(), left.getQualifierOffset(), lqlength, right.getQualifierArray(), right.getQualifierOffset(), rqlength); }
@Override public void internalEncodeKeyValues(DataOutputStream out, ByteBuffer in, boolean includesMemstoreTS) throws IOException { in.rewind(); ByteBufferUtils.putInt(out, in.limit()); DiffCompressionState previousState = new DiffCompressionState(); DiffCompressionState currentState = new DiffCompressionState(); while (in.hasRemaining()) { compressSingleKeyValue(previousState, currentState, out, in); afterEncodingKeyValue(in, out, includesMemstoreTS); // swap previousState <-> currentState DiffCompressionState tmp = previousState; previousState = currentState; currentState = tmp; } }
public static boolean matchingFamily(final Cell left, final Cell right) { byte lfamlength = left.getFamilyLength(); byte rfamlength = right.getFamilyLength(); if (left instanceof ByteBufferExtendedCell && right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getFamilyByteBuffer(), ((ByteBufferExtendedCell) left).getFamilyPosition(), lfamlength, ((ByteBufferExtendedCell) right).getFamilyByteBuffer(), ((ByteBufferExtendedCell) right).getFamilyPosition(), rfamlength); } if (left instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getFamilyByteBuffer(), ((ByteBufferExtendedCell) left).getFamilyPosition(), lfamlength, right.getFamilyArray(), right.getFamilyOffset(), rfamlength); } if (right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) right).getFamilyByteBuffer(), ((ByteBufferExtendedCell) right).getFamilyPosition(), rfamlength, left.getFamilyArray(), left.getFamilyOffset(), lfamlength); } return Bytes.equals(left.getFamilyArray(), left.getFamilyOffset(), lfamlength, right.getFamilyArray(), right.getFamilyOffset(), rfamlength); }
private short getShort(int index, int itemIndex) { ByteBuffer item = items[itemIndex]; int offsetInItem = index - this.itemBeginPos[itemIndex]; int remainingLen = item.limit() - offsetInItem; if (remainingLen >= Bytes.SIZEOF_SHORT) { return ByteBufferUtils.toShort(item, offsetInItem); } if (items.length - 1 == itemIndex) { // means cur item is the last one and we wont be able to read a int. Throw exception throw new BufferUnderflowException(); } ByteBuffer nextItem = items[itemIndex + 1]; // Get available bytes from this item and remaining from next short l = 0; for (int i = offsetInItem; i < item.capacity(); i++) { l = (short) (l << 8); l = (short) (l ^ (ByteBufferUtils.toByte(item, i) & 0xFF)); } for (int i = 0; i < Bytes.SIZEOF_SHORT - remainingLen; i++) { l = (short) (l << 8); l = (short) (l ^ (ByteBufferUtils.toByte(item, i) & 0xFF)); } return l; }
@Override public SingleByteBuff put(int offset, ByteBuff src, int srcOffset, int length) { if (src instanceof SingleByteBuff) { ByteBufferUtils.copyFromBufferToBuffer(((SingleByteBuff) src).buf, this.buf, srcOffset, offset, length); } else { // TODO we can do some optimization here? Call to asSubByteBuffer might // create a copy. ObjectIntPair<ByteBuffer> pair = new ObjectIntPair<>(); src.asSubByteBuffer(srcOffset, length, pair); if (pair.getFirst() != null) { ByteBufferUtils.copyFromBufferToBuffer(pair.getFirst(), this.buf, pair.getSecond(), offset, length); } } return this; }
@Override public void get(int sourceOffset, byte[] dst, int offset, int length) { int itemIndex = getItemIndex(sourceOffset); ByteBuffer item = this.items[itemIndex]; sourceOffset = sourceOffset - this.itemBeginPos[itemIndex]; while (length > 0) { int toRead = Math.min((item.limit() - sourceOffset), length); ByteBufferUtils.copyFromBufferToArray(dst, item, sourceOffset, offset, toRead); length -= toRead; if (length == 0) break; itemIndex++; item = this.items[itemIndex]; offset += toRead; sourceOffset = 0; } }
/** * Retrieve Cell's first tag, matching the passed in type * @param cell The Cell * @param type Type of the Tag to retrieve * @return null if there is no tag of the passed in tag type */ public static Optional<Tag> getTag(Cell cell, byte type) { boolean bufferBacked = cell instanceof ByteBufferExtendedCell; int length = cell.getTagsLength(); int offset = bufferBacked ? ((ByteBufferExtendedCell) cell).getTagsPosition() : cell.getTagsOffset(); int pos = offset; while (pos < offset + length) { int tagLen; if (bufferBacked) { ByteBuffer tagsBuffer = ((ByteBufferExtendedCell) cell).getTagsByteBuffer(); tagLen = ByteBufferUtils.readAsInt(tagsBuffer, pos, TAG_LENGTH_SIZE); if (ByteBufferUtils.toByte(tagsBuffer, pos + TAG_LENGTH_SIZE) == type) { return Optional.of(new ByteBufferTag(tagsBuffer, pos, tagLen + TAG_LENGTH_SIZE)); } } else { tagLen = Bytes.readAsInt(cell.getTagsArray(), pos, TAG_LENGTH_SIZE); if (cell.getTagsArray()[pos + TAG_LENGTH_SIZE] == type) { return Optional .of(new ArrayBackedTag(cell.getTagsArray(), pos, tagLen + TAG_LENGTH_SIZE)); } } pos += TAG_LENGTH_SIZE + tagLen; } return Optional.empty(); }
private int getInt(int index, int itemIndex) { ByteBuffer item = items[itemIndex]; int offsetInItem = index - this.itemBeginPos[itemIndex]; int remainingLen = item.limit() - offsetInItem; if (remainingLen >= Bytes.SIZEOF_INT) { return ByteBufferUtils.toInt(item, offsetInItem); } if (items.length - 1 == itemIndex) { // means cur item is the last one and we wont be able to read a int. Throw exception throw new BufferUnderflowException(); } ByteBuffer nextItem = items[itemIndex + 1]; // Get available bytes from this item and remaining from next int l = 0; for (int i = offsetInItem; i < item.capacity(); i++) { l <<= 8; l ^= ByteBufferUtils.toByte(item, i) & 0xFF; } for (int i = 0; i < Bytes.SIZEOF_INT - remainingLen; i++) { l <<= 8; l ^= ByteBufferUtils.toByte(nextItem, i) & 0xFF; } return l; }
/** * Writes the cell to the given OutputStream * @param cell the cell to be written * @param out the outputstream * @param withTags if tags are to be written or not * @return the total bytes written * @throws IOException */ public static int writeCell(Cell cell, OutputStream out, boolean withTags) throws IOException { if (cell instanceof ExtendedCell) { return ((ExtendedCell) cell).write(out, withTags); } else { ByteBufferUtils.putInt(out, estimatedSerializedSizeOfKey(cell)); ByteBufferUtils.putInt(out, cell.getValueLength()); writeFlatKey(cell, out); writeValue(out, cell, cell.getValueLength()); int tagsLength = cell.getTagsLength(); if (withTags) { byte[] len = new byte[Bytes.SIZEOF_SHORT]; Bytes.putAsShort(len, 0, tagsLength); out.write(len); if (tagsLength > 0) { writeTags(out, cell, tagsLength); } } int lenWritten = (2 * Bytes.SIZEOF_INT) + estimatedSerializedSizeOfKey(cell) + cell.getValueLength(); if (withTags) { lenWritten += Bytes.SIZEOF_SHORT + tagsLength; } return lenWritten; } }
public static boolean matchingValue(final Cell left, final Cell right, int lvlength, int rvlength) { if (left instanceof ByteBufferExtendedCell && right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getValueByteBuffer(), ((ByteBufferExtendedCell) left).getValuePosition(), lvlength, ((ByteBufferExtendedCell) right).getValueByteBuffer(), ((ByteBufferExtendedCell) right).getValuePosition(), rvlength); } if (left instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getValueByteBuffer(), ((ByteBufferExtendedCell) left).getValuePosition(), lvlength, right.getValueArray(), right.getValueOffset(), rvlength); } if (right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) right).getValueByteBuffer(), ((ByteBufferExtendedCell) right).getValuePosition(), rvlength, left.getValueArray(), left.getValueOffset(), lvlength); } return Bytes.equals(left.getValueArray(), left.getValueOffset(), lvlength, right.getValueArray(), right.getValueOffset(), rvlength); }
@Override public int write(OutputStream out, boolean withTags) throws IOException { int lenToWrite = getSerializedSize(withTags); ByteBufferUtils.putInt(out, keyOnlyBuffer.length); ByteBufferUtils.putInt(out, valueLength); // Write key out.write(keyOnlyBuffer); // Write value out.write(this.valueBuffer, this.valueOffset, this.valueLength); if (withTags && this.tagsLength > 0) { // 2 bytes tags length followed by tags bytes // tags length is serialized with 2 bytes only(short way) even if the type is int. // As this is non -ve numbers, we save the sign bit. See HBASE-11437 out.write((byte) (0xff & (this.tagsLength >> 8))); out.write((byte) (0xff & this.tagsLength)); out.write(this.tagsBuffer, this.tagsOffset, this.tagsLength); } return lenToWrite; }
@Override public void internalEncodeKeyValues(DataOutputStream out, ByteBuffer in, boolean includesMemstoreTS) throws IOException { in.rewind(); ByteBufferUtils.putInt(out, in.limit()); FastDiffCompressionState previousState = new FastDiffCompressionState(); FastDiffCompressionState currentState = new FastDiffCompressionState(); while (in.hasRemaining()) { compressSingleKeyValue(previousState, currentState, out, in); afterEncodingKeyValue(in, out, includesMemstoreTS); // swap previousState <-> currentState FastDiffCompressionState tmp = previousState; previousState = currentState; currentState = tmp; } }
/** * Compares the row of two keyvalues for equality * @param left * @param right * @return True if rows match. */ public static boolean matchingRows(final Cell left, final Cell right) { short lrowlength = left.getRowLength(); short rrowlength = right.getRowLength(); if (lrowlength != rrowlength) return false; if (left instanceof ByteBufferExtendedCell && right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getRowByteBuffer(), ((ByteBufferExtendedCell) left).getRowPosition(), lrowlength, ((ByteBufferExtendedCell) right).getRowByteBuffer(), ((ByteBufferExtendedCell) right).getRowPosition(), rrowlength); } if (left instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getRowByteBuffer(), ((ByteBufferExtendedCell) left).getRowPosition(), lrowlength, right.getRowArray(), right.getRowOffset(), rrowlength); } if (right instanceof ByteBufferExtendedCell) { return ByteBufferUtils.equals(((ByteBufferExtendedCell) right).getRowByteBuffer(), ((ByteBufferExtendedCell) right).getRowPosition(), rrowlength, left.getRowArray(), left.getRowOffset(), lrowlength); } return Bytes.equals(left.getRowArray(), left.getRowOffset(), lrowlength, right.getRowArray(), right.getRowOffset(), rrowlength); }
/** * An approximation to the {@link HFile}'s mid-key. Operates on block * boundaries, and does not go inside blocks. In other words, returns the * first key of the middle block of the file. * * @return the first key of the middle block */ public byte[] midkey() throws IOException { if (rootCount == 0) throw new IOException("HFile empty"); byte[] targetMidKey = this.midKey.get(); if (targetMidKey != null) { return targetMidKey; } if (midLeafBlockOffset >= 0) { if (cachingBlockReader == null) { throw new IOException("Have to read the middle leaf block but " + "no block reader available"); } // Caching, using pread, assuming this is not a compaction. HFileBlock midLeafBlock = cachingBlockReader.readBlock( midLeafBlockOffset, midLeafBlockOnDiskSize, true, true, false, true, BlockType.LEAF_INDEX, null); ByteBuffer b = midLeafBlock.getBufferWithoutHeader(); int numDataBlocks = b.getInt(); int keyRelOffset = b.getInt(Bytes.SIZEOF_INT * (midKeyEntry + 1)); int keyLen = b.getInt(Bytes.SIZEOF_INT * (midKeyEntry + 2)) - keyRelOffset; int keyOffset = Bytes.SIZEOF_INT * (numDataBlocks + 2) + keyRelOffset + SECONDARY_INDEX_ENTRY_OVERHEAD; targetMidKey = ByteBufferUtils.toBytes(b, keyOffset, keyLen); } else { // The middle of the root-level index. targetMidKey = blockKeys[rootCount / 2]; } this.midKey.set(targetMidKey); return targetMidKey; }
/** * The indexed key at the ith position in the nonRootIndex. The position starts at 0. * @param nonRootIndex * @param i the ith position * @return The indexed key at the ith position in the nonRootIndex. */ private byte[] getNonRootIndexedKey(ByteBuffer nonRootIndex, int i) { int numEntries = nonRootIndex.getInt(0); if (i < 0 || i >= numEntries) { return null; } // Entries start after the number of entries and the secondary index. // The secondary index takes numEntries + 1 ints. int entriesOffset = Bytes.SIZEOF_INT * (numEntries + 2); // Targetkey's offset relative to the end of secondary index int targetKeyRelOffset = nonRootIndex.getInt( Bytes.SIZEOF_INT * (i + 1)); // The offset of the target key in the blockIndex buffer int targetKeyOffset = entriesOffset // Skip secondary index + targetKeyRelOffset // Skip all entries until mid + SECONDARY_INDEX_ENTRY_OVERHEAD; // Skip offset and on-disk-size // We subtract the two consecutive secondary index elements, which // gives us the size of the whole (offset, onDiskSize, key) tuple. We // then need to subtract the overhead of offset and onDiskSize. int targetKeyLength = nonRootIndex.getInt(Bytes.SIZEOF_INT * (i + 2)) - targetKeyRelOffset - SECONDARY_INDEX_ENTRY_OVERHEAD; return ByteBufferUtils.toBytes(nonRootIndex, targetKeyOffset, targetKeyLength); }
@Override public boolean equals(Object comparison) { if (this == comparison) { return true; } if (comparison == null) { return false; } if (comparison.getClass() != this.getClass()) { return false; } HFileBlock castedComparison = (HFileBlock) comparison; if (castedComparison.blockType != this.blockType) { return false; } if (castedComparison.nextBlockOnDiskSizeWithHeader != this.nextBlockOnDiskSizeWithHeader) { return false; } if (castedComparison.offset != this.offset) { return false; } if (castedComparison.onDiskSizeWithoutHeader != this.onDiskSizeWithoutHeader) { return false; } if (castedComparison.prevBlockOffset != this.prevBlockOffset) { return false; } if (castedComparison.uncompressedSizeWithoutHeader != this.uncompressedSizeWithoutHeader) { return false; } if (ByteBufferUtils.compareTo(this.buf, 0, this.buf.limit(), castedComparison.buf, 0, castedComparison.buf.limit()) != 0) { return false; } return true; }
/** * I don't think this method is called during normal HBase operation, so efficiency is not * important. */ public ByteBuffer decodeKeyValues(DataInputStream source, int allocateHeaderLength, int skipLastBytes, HFileBlockDecodingContext decodingCtx) throws IOException { ByteBuffer sourceAsBuffer = ByteBufferUtils.drainInputStreamToBuffer(source);// waste sourceAsBuffer.mark(); PrefixTreeBlockMeta blockMeta = new PrefixTreeBlockMeta(sourceAsBuffer); sourceAsBuffer.rewind(); int numV1BytesWithHeader = allocateHeaderLength + blockMeta.getNumKeyValueBytes(); byte[] keyValueBytesWithHeader = new byte[numV1BytesWithHeader]; ByteBuffer result = ByteBuffer.wrap(keyValueBytesWithHeader); result.rewind(); CellSearcher searcher = null; try { boolean includesMvcc = decodingCtx.getHFileContext().isIncludesMvcc(); searcher = DecoderFactory.checkOut(sourceAsBuffer, includesMvcc); while (searcher.advance()) { KeyValue currentCell = KeyValueUtil.copyToNewKeyValue(searcher.current()); // needs to be modified for DirectByteBuffers. no existing methods to // write VLongs to byte[] int offset = result.arrayOffset() + result.position(); System.arraycopy(currentCell.getBuffer(), currentCell.getOffset(), result.array(), offset, currentCell.getLength()); int keyValueLength = KeyValueUtil.length(currentCell); ByteBufferUtils.skip(result, keyValueLength); offset += keyValueLength; if (includesMvcc) { ByteBufferUtils.writeVLong(result, currentCell.getMvccVersion()); } } result.position(result.limit());//make it appear as if we were appending return result; } finally { DecoderFactory.checkIn(searcher); } }
public static void appendToByteBuffer(final ByteBuffer bb, final KeyValue kv, final boolean includeMvccVersion) { // keep pushing the limit out. assume enough capacity bb.limit(bb.position() + kv.getLength()); bb.put(kv.getBuffer(), kv.getOffset(), kv.getLength()); if (includeMvccVersion) { int numMvccVersionBytes = WritableUtils.getVIntSize(kv.getMvccVersion()); ByteBufferUtils.extendLimit(bb, numMvccVersionBytes); ByteBufferUtils.writeVLong(bb, kv.getMvccVersion()); } }
/** * Creates a new KeyValue object positioned in the supplied ByteBuffer and sets the ByteBuffer's * position to the start of the next KeyValue. Does not allocate a new array or copy data. * @param bb * @param includesMvccVersion * @param includesTags */ public static KeyValue nextShallowCopy(final ByteBuffer bb, final boolean includesMvccVersion, boolean includesTags) { if (bb.isDirect()) { throw new IllegalArgumentException("only supports heap buffers"); } if (bb.remaining() < 1) { return null; } KeyValue keyValue = null; int underlyingArrayOffset = bb.arrayOffset() + bb.position(); int keyLength = bb.getInt(); int valueLength = bb.getInt(); ByteBufferUtils.skip(bb, keyLength + valueLength); int tagsLength = 0; if (includesTags) { // Read short as unsigned, high byte first tagsLength = ((bb.get() & 0xff) << 8) ^ (bb.get() & 0xff); ByteBufferUtils.skip(bb, tagsLength); } int kvLength = (int) KeyValue.getKeyValueDataStructureSize(keyLength, valueLength, tagsLength); keyValue = new KeyValue(bb.array(), underlyingArrayOffset, kvLength); if (includesMvccVersion) { long mvccVersion = ByteBufferUtils.readVLong(bb); keyValue.setSequenceId(mvccVersion); } return keyValue; }
/** * Compress tags one by one and writes to the OutputStream. * @param out Stream to which the compressed tags to be written * @param in Source buffer where tags are available * @param length Length of all tag bytes * @throws IOException */ public void compressTags(OutputStream out, ByteBuffer in, int length) throws IOException { if (in.hasArray()) { compressTags(out, in.array(), in.arrayOffset() + in.position(), length); ByteBufferUtils.skip(in, length); } else { byte[] tagBuf = new byte[length]; in.get(tagBuf); compressTags(out, tagBuf, 0, length); } }
@Override public EncodedSeeker createSeeker(KVComparator comparator, final HFileBlockDecodingContext decodingCtx) { return new BufferedEncodedSeeker<SeekerState>(comparator, decodingCtx) { @Override protected void decodeNext() { current.keyLength = currentBuffer.getInt(); current.valueLength = currentBuffer.getInt(); current.ensureSpaceForKey(); currentBuffer.get(current.keyBuffer, 0, current.keyLength); current.valueOffset = currentBuffer.position(); ByteBufferUtils.skip(currentBuffer, current.valueLength); if (includesTags()) { // Read short as unsigned, high byte first current.tagsLength = ((currentBuffer.get() & 0xff) << 8) ^ (currentBuffer.get() & 0xff); ByteBufferUtils.skip(currentBuffer, current.tagsLength); } if (includesMvcc()) { current.memstoreTS = ByteBufferUtils.readVLong(currentBuffer); } else { current.memstoreTS = 0; } current.nextKvOffset = currentBuffer.position(); } @Override protected void decodeFirst() { ByteBufferUtils.skip(currentBuffer, Bytes.SIZEOF_INT); current.lastCommonPrefix = 0; decodeNext(); } }; }