.../ipc/TestPayloadCarryingRpcController.java | 6 + .../apache/hadoop/hbase/ByteBufferBackedCell.java | 71 +++ .../hadoop/hbase/ByteBufferBackedKeyValue.java | 518 +++++++++++++++++++++ .../hadoop/hbase/ByteBufferBackedKeyValueUtil.java | 196 ++++++++ .../main/java/org/apache/hadoop/hbase/Cell.java | 8 +- .../java/org/apache/hadoop/hbase/KeyValue.java | 5 + .../src/main/java/org/apache/hadoop/hbase/Tag.java | 6 + .../io/encoding/BufferedDataBlockEncoder.java | 10 + .../apache/hadoop/hbase/util/ByteBufferUtils.java | 513 +++++++++++++++++++- .../java/org/apache/hadoop/hbase/TestCellUtil.java | 8 +- .../hbase/codec/prefixtree/PrefixTreeSeeker.java | 5 + .../codec/prefixtree/decode/PrefixTreeCell.java | 5 + .../org/apache/hadoop/hbase/TagRewriteCell.java | 5 + 13 files changed, 1350 insertions(+), 6 deletions(-) diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/ipc/TestPayloadCarryingRpcController.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/ipc/TestPayloadCarryingRpcController.java index e6d6f43..7c23cde 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/ipc/TestPayloadCarryingRpcController.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/ipc/TestPayloadCarryingRpcController.java @@ -206,6 +206,12 @@ public class TestPayloadCarryingRpcController { // unused return null; } + + @Override + public boolean hasArray() { + // TODO Auto-generated method stub + return true; + } }; } diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedCell.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedCell.java new file mode 100644 index 0000000..7a1e543 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedCell.java @@ -0,0 +1,71 @@ +/* + * 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; + +import java.nio.ByteBuffer; + +import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.io.HeapSize; +/** + * An extension of the {@link Cell} interface that provides API + * to access the parts of the cell (row, family, qualifier, value, tags) + * as ByteBuffer. Used in the read path on the server side. + */ +@InterfaceAudience.Private +public interface ByteBufferBackedCell extends Cell, SettableSequenceId, HeapSize { + + /** + * ByteBuffer containing the row bytes. + * Max length is Short.MAX_VALUE which is 32,767 bytes. + * + * @return The ByteBuffer containing the row bytes. + */ + ByteBuffer getRowBuffer(); + + /** + * ByteBuffer containing the family bytes. + * Max length is Byte.MAX_VALUE which is 127 bytes. + * + * @return the ByteBuffer containing the family bytes. + */ + ByteBuffer getFamilyBuffer(); + + /** + * ByteBuffer containing the qualifier bytes. + * Max length is Short.MAX_VALUE which is 32,767 bytes. + * + * @return The ByteBuffer containing the qualifier bytes. + */ + ByteBuffer getQualifierBuffer(); + + /** + * ByteBuffer containing the value bytes. + * Max length is Integer.MAX_VALUE which is 2,147,483,648 bytes. + * + * @return The ByteBuffer containing the value bytes. + */ + ByteBuffer getValueBuffer(); + + /** + * ByteBuffer containing the tag bytes. + * Max length is Integer.MAX_VALUE which is 2,147,483,648 bytes. + * + * @return The ByteBuffer containing the tag bytes. + */ + ByteBuffer getTagsBuffer(); +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedKeyValue.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedKeyValue.java new file mode 100644 index 0000000..ec9a5ff --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedKeyValue.java @@ -0,0 +1,518 @@ +/* + * 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; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; + +import org.apache.hadoop.hbase.KeyValue.Type; +import org.apache.hadoop.hbase.io.HeapSize; +import org.apache.hadoop.hbase.util.ByteBufferUtils; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.ClassSize; + +public class ByteBufferBackedKeyValue implements ByteBufferBackedCell, HeapSize, Cloneable, + SettableSequenceId { + /** Size of the key length field in bytes */ + public static final int KEY_LENGTH_SIZE = Bytes.SIZEOF_INT; + + /** Size of the key type field in bytes */ + public static final int TYPE_SIZE = Bytes.SIZEOF_BYTE; + + /** Size of the row length field in bytes */ + public static final int ROW_LENGTH_SIZE = Bytes.SIZEOF_SHORT; + + /** Size of the family length field in bytes */ + public static final int FAMILY_LENGTH_SIZE = Bytes.SIZEOF_BYTE; + + /** Size of the timestamp field in bytes */ + public static final int TIMESTAMP_SIZE = Bytes.SIZEOF_LONG; + + // Size of the timestamp and type byte on end of a key -- a long + a byte. + public static final int TIMESTAMP_TYPE_SIZE = TIMESTAMP_SIZE + TYPE_SIZE; + + // Size of the length shorts and bytes in key. + public static final int KEY_INFRASTRUCTURE_SIZE = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + + TIMESTAMP_TYPE_SIZE; + + // How far into the key the row starts at. First thing to read is the short + // that says how long the row is. + public static final int ROW_OFFSET = Bytes.SIZEOF_INT /* keylength */+ Bytes.SIZEOF_INT /* valuelength */; + + // Size of the length ints in a KeyValue datastructure. + public static final int KEYVALUE_INFRASTRUCTURE_SIZE = ROW_OFFSET; + + /** Size of the tags length field in bytes */ + public static final int TAGS_LENGTH_SIZE = Bytes.SIZEOF_SHORT; + + public static final int KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE = ROW_OFFSET + TAGS_LENGTH_SIZE; + + private static final int MAX_TAGS_LENGTH = (2 * Short.MAX_VALUE) + 1; + + private ByteBuffer buffer; + private int offset; + private int length; + + public ByteBufferBackedKeyValue() { + + } + + public ByteBufferBackedKeyValue(ByteBuffer buffer, int offset, int length) { + this.buffer = buffer; + // The offset that we set here from the HFileReader should not have the + // arrayOffset. + // All the ByteBufferUtils API should use it internally. + this.offset = offset; + this.length = length; + } + + public ByteBufferBackedKeyValue(final ByteBuffer row, final ByteBuffer family, + final ByteBuffer qualifier, final long timestamp, Type type) { + this(row, family, qualifier, timestamp, type, null); + } + + public ByteBufferBackedKeyValue(final ByteBuffer row, final ByteBuffer family, + final ByteBuffer qualifier, final long timestamp, Type type, final ByteBuffer value) { + this(row, 0, len(row), family, 0, len(family), qualifier, 0, len(qualifier), timestamp, type, + value, 0, len(value)); + } + + public ByteBufferBackedKeyValue(final ByteBuffer row, final ByteBuffer family, + final ByteBuffer qualifier, final long timestamp, Type type, final ByteBuffer value, + final List tags) { + this(row, family, qualifier, 0, qualifier == null ? 0 : qualifier.capacity(), timestamp, type, + value, 0, value == null ? 0 : value.capacity(), tags); + } + + public ByteBufferBackedKeyValue(ByteBuffer row, ByteBuffer family, ByteBuffer qualifier, + int qoffset, int qlength, long timestamp, Type type, ByteBuffer value, int voffset, + int vlength, List tags) { + this(row, 0, row == null ? 0 : row.capacity(), family, 0, family == null ? 0 : family + .capacity(), qualifier, qoffset, qlength, timestamp, type, value, voffset, vlength, tags); + } + + final public static int len(ByteBuffer b) { + return b == null ? 0 : b.capacity(); + } + + public ByteBufferBackedKeyValue(final ByteBuffer row, final int roffset, final int rlength, + final ByteBuffer family, final int foffset, final int flength, final ByteBuffer qualifier, + final int qoffset, final int qlength, final long timestamp, final Type type, + final ByteBuffer value, final int voffset, final int vlength) { + this(row, roffset, rlength, family, foffset, flength, qualifier, qoffset, qlength, timestamp, + type, value, voffset, vlength, null); + } + + public ByteBufferBackedKeyValue(final ByteBuffer row, final int roffset, final int rlength, + final ByteBuffer family, final int foffset, final int flength, final ByteBuffer qualifier, + final int qoffset, final int qlength, final long timestamp, final Type type, + final ByteBuffer value, final int voffset, final int vlength, final List tags) { + this.buffer = createByteArray(row, roffset, rlength, family, foffset, flength, qualifier, + qoffset, qlength, timestamp, type, value, voffset, vlength, tags); + this.length = buffer.capacity(); + this.offset = 0; + } + + private static ByteBuffer createByteArray(final ByteBuffer row, final int roffset, + final int rlength, final ByteBuffer family, final int foffset, int flength, + final ByteBuffer qualifier, final int qoffset, int qlength, final long timestamp, + final Type type, final ByteBuffer value, final int voffset, int vlength, List tags) { + + checkParameters(row, rlength, family, flength, qlength, vlength); + + // Calculate length of tags area + int tagsLength = 0; + if (tags != null && !tags.isEmpty()) { + for (Tag t : tags) { + tagsLength += t.getLength(); + } + } + checkForTagsLength(tagsLength); + // Allocate right-sized byte array. + int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength); + ByteBuffer buffer = ByteBuffer.allocate((int) getKeyValueDataStructureSize(rlength, flength, + qlength, vlength, tagsLength)); + + // Write key, value and key row length. + int pos = 0; + pos = ByteBufferUtils.putInt(buffer, pos, keyLength); + + pos = ByteBufferUtils.putInt(buffer, pos, vlength); + pos = ByteBufferUtils.putShort(buffer, pos, (short) (rlength & 0x0000ffff)); + pos = ByteBufferUtils.copyFromBufferToBuffer(buffer, row, roffset, pos, rlength); + pos = ByteBufferUtils.putByte(buffer, pos, (byte) (flength & 0x0000ff)); + if (flength != 0) { + pos = ByteBufferUtils.copyFromBufferToBuffer(buffer, family, foffset, pos, flength); + } + if (qlength > 0) { + pos = ByteBufferUtils.copyFromBufferToBuffer(buffer, qualifier, qoffset, pos, qlength); + } + pos = ByteBufferUtils.putLong(buffer, pos, timestamp); + pos = ByteBufferUtils.putByte(buffer, pos, type.getCode()); + if (vlength > 0) { + pos = ByteBufferUtils.copyFromBufferToBuffer(buffer, value, voffset, pos, vlength); + } + // Add the tags after the value part + if (tagsLength > 0) { + pos = ByteBufferUtils.putShort(buffer, pos, tagsLength); + for (Tag t : tags) { + pos = ByteBufferUtils.copyFromBufferToBuffer(buffer, t.getAsByteBuffer(), t.getOffset(), + pos, t.getLength()); + } + } + return buffer; + } + + public static long getKeyValueDataStructureSize(int rlength, int flength, int qlength, + int vlength, int tagsLength) { + if (tagsLength == 0) { + return getKeyValueDataStructureSize(rlength, flength, qlength, vlength); + } + return KeyValue.KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE + + getKeyDataStructureSize(rlength, flength, qlength) + vlength + tagsLength; + } + + public static long getKeyValueDataStructureSize(int rlength, int flength, int qlength, int vlength) { + return KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + + getKeyDataStructureSize(rlength, flength, qlength) + vlength; + } + + private static void checkParameters(final ByteBuffer row, final int rlength, + final ByteBuffer family, int flength, int qlength, int vlength) + throws IllegalArgumentException { + if (rlength > Short.MAX_VALUE) { + throw new IllegalArgumentException("Row > " + Short.MAX_VALUE); + } + if (row == null) { + throw new IllegalArgumentException("Row is null"); + } + // Family length + flength = family == null ? 0 : flength; + if (flength > Byte.MAX_VALUE) { + throw new IllegalArgumentException("Family > " + Byte.MAX_VALUE); + } + // Qualifier length + if (qlength > Integer.MAX_VALUE - rlength - flength) { + throw new IllegalArgumentException("Qualifier > " + Integer.MAX_VALUE); + } + // Key length + long longKeyLength = getKeyDataStructureSize(rlength, flength, qlength); + if (longKeyLength > Integer.MAX_VALUE) { + throw new IllegalArgumentException("keylength " + longKeyLength + " > " + Integer.MAX_VALUE); + } + // Value length + if (vlength > HConstants.MAXIMUM_VALUE_LENGTH) { // FindBugs + // INT_VACUOUS_COMPARISON + throw new IllegalArgumentException("Value length " + vlength + " > " + + HConstants.MAXIMUM_VALUE_LENGTH); + } + } + + private static void checkForTagsLength(int tagsLength) { + if (tagsLength > MAX_TAGS_LENGTH) { + throw new IllegalArgumentException("tagslength " + tagsLength + " > " + MAX_TAGS_LENGTH); + } + } + + @Override + public byte[] getRowArray() { + // TODO : Do some work around if we should not throw + // UnsupportedOperationException + throw new UnsupportedOperationException("byte[] not supported"); + } + + @Override + public int getRowOffset() { + return getKeyOffset() + Bytes.SIZEOF_SHORT; + } + + @Override + public short getRowLength() { + return ByteBufferUtils.readAsShort(this.buffer, getKeyOffset()); + } + + @Override + public byte[] getFamilyArray() { + // TODO : Do some work around if we should not throw + // UnsupportedOperationException + throw new UnsupportedOperationException("byte[] not supported"); + } + + @Override + public int getFamilyOffset() { + return getFamilyOffset(getRowLength()); + } + + private int getFamilyOffset(int rlength) { + return this.offset + ROW_OFFSET + Bytes.SIZEOF_SHORT + rlength + Bytes.SIZEOF_BYTE; + } + + @Override + public byte getFamilyLength() { + return getFamilyLength(getFamilyOffset()); + } + + private byte getFamilyLength(int foffset) { + return this.buffer.get(foffset - 1); + } + + @Override + public byte[] getQualifierArray() { + // TODO : Do some work around if we should not throw + // UnsupportedOperationException + throw new UnsupportedOperationException("byte[] not supported"); + } + + @Override + public int getQualifierOffset() { + return getQualifierOffset(getFamilyOffset()); + } + + private int getQualifierOffset(int foffset) { + return foffset + getFamilyLength(foffset); + } + + @Override + public int getQualifierLength() { + return getQualifierLength(getRowLength(), getFamilyLength()); + } + + private int getQualifierLength(int rlength, int flength) { + return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0); + } + + public static long getKeyDataStructureSize(int rlength, int flength, int qlength) { + return KeyValue.KEY_INFRASTRUCTURE_SIZE + rlength + flength + qlength; + } + + @Override + public long getTimestamp() { + return getTimestampOffset(getKeyLength()); + } + + /** + * @param keylength + * Pass if you have it to save on a int creation. + * @return Timestamp offset + */ + private int getTimestampOffset(final int keylength) { + return getKeyOffset() + keylength - TIMESTAMP_TYPE_SIZE; + } + + @Override + public byte getTypeByte() { + return ByteBufferUtils.readAsByte(this.buffer, this.offset + getKeyLength() - 1 + ROW_OFFSET); + } + + @Override + @Deprecated + public long getMvccVersion() { + // TODO Auto-generated method stub + return 0; + } + + @Override + public long getSequenceId() { + return seqId; + } + + private long seqId = 0; + + @Override + public byte[] getValueArray() { + // TODO : Do some work around if we should not throw + // UnsupportedOperationException + throw new UnsupportedOperationException("byte[] not supported"); + } + + @Override + public int getValueOffset() { + int voffset = getKeyOffset() + getKeyLength(); + return voffset; + } + + @Override + public int getValueLength() { + int vlength = ByteBufferUtils.readAsInt(this.buffer, this.offset + Bytes.SIZEOF_INT); + return vlength; + } + + @Override + public byte[] getTagsArray() { + // TODO : Do some work around if we should not throw + // UnsupportedOperationException + throw new UnsupportedOperationException("byte[] not supported"); + } + + @Override + public int getTagsOffset() { + int tagsLen = getTagsLength(); + if (tagsLen == 0) { + return this.offset + this.length; + } + return this.offset + this.length - tagsLen; + } + + @Override + public int getTagsLength() { + int tagsLen = this.length - (getKeyLength() + getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE); + if (tagsLen > 0) { + // There are some Tag bytes in the byte[]. So reduce 2 bytes which is + // added to denote the tags + // length + tagsLen -= TAGS_LENGTH_SIZE; + } + return tagsLen; + } + + @Override + public boolean hasArray() { + return false; + } + + @Override + @Deprecated + public byte[] getValue() { + // TODO Auto-generated method stub + return null; + } + + @Override + @Deprecated + public byte[] getFamily() { + // TODO Auto-generated method stub + return null; + } + + @Override + @Deprecated + public byte[] getQualifier() { + // TODO Auto-generated method stub + return null; + } + + @Override + @Deprecated + public byte[] getRow() { + // TODO Auto-generated method stub + return null; + } + + @Override + public void setSequenceId(long seqId) throws IOException { + this.seqId = seqId; + } + + @Override + public long heapSize() { + int sum = 0; + sum += ClassSize.OBJECT;// the KeyValue object itself + sum += ClassSize.REFERENCE;// pointer to "buffer" + sum += ClassSize.align(length);// number of bytes of data in the "bytes" + sum += 2 * Bytes.SIZEOF_INT;// offset, length + sum += Bytes.SIZEOF_LONG;// memstoreTS + return ClassSize.align(sum); + } + + @Override + public ByteBuffer getRowBuffer() { + return buffer; + } + + @Override + public ByteBuffer getFamilyBuffer() { + return buffer; + } + + @Override + public ByteBuffer getQualifierBuffer() { + return buffer; + } + + @Override + public ByteBuffer getValueBuffer() { + return buffer; + } + + @Override + public ByteBuffer getTagsBuffer() { + return buffer; + } + + public int getKeyOffset() { + return this.offset + ROW_OFFSET; + } + + /** + * @return Length of key portion. + */ + public int getKeyLength() { + return ByteBufferUtils.readAsInt(this.buffer, this.offset); + } + + public ByteBuffer getKey() { + // TODO : Need test cases for this + ByteBuffer dup = getBuffer().duplicate(); + dup.position(getKeyOffset()); + dup.limit(getKeyLength()); + return dup.slice(); + } + + public ByteBuffer getBuffer() { + return this.buffer; + } + + @Override + public String toString() { + if (this.buffer == null || this.buffer.capacity() == 0) { + return "empty"; + } + return keyToString(this.buffer, this.offset + ROW_OFFSET, getKeyLength()) + "/vlen=" + + getValueLength() + "/seqid=" + seqId; + } + + public static String keyToString(final ByteBuffer b, final int o, final int l) { + if (b == null) + return ""; + int rowlength = ByteBufferUtils.readAsShort(b, o); + String row = ByteBufferUtils.toStringBinary(b, o + Bytes.SIZEOF_SHORT, rowlength); + int columnoffset = o + Bytes.SIZEOF_SHORT + 1 + rowlength; + int familylength = b.get(columnoffset - 1); + int columnlength = l - ((columnoffset - o) + TIMESTAMP_TYPE_SIZE); + String family = familylength == 0 ? "" : ByteBufferUtils.toStringBinary(b, columnoffset, + familylength); + String qualifier = columnlength == 0 ? "" : ByteBufferUtils.toStringBinary(b, columnoffset + + familylength, columnlength - familylength); + long timestamp = ByteBufferUtils.readAsLong(b, o + (l - TIMESTAMP_TYPE_SIZE)); + String timestampStr = humanReadableTimestamp(timestamp); + byte type = b.get(o + l - 1); + return row + "/" + family + (family != null && family.length() > 0 ? ":" : "") + qualifier + + "/" + timestampStr + "/" + Type.codeToType(type); + } + + public static String humanReadableTimestamp(final long timestamp) { + if (timestamp == HConstants.LATEST_TIMESTAMP) { + return "LATEST_TIMESTAMP"; + } + if (timestamp == HConstants.OLDEST_TIMESTAMP) { + return "OLDEST_TIMESTAMP"; + } + return String.valueOf(timestamp); + } +} \ No newline at end of file diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedKeyValueUtil.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedKeyValueUtil.java new file mode 100644 index 0000000..98ef3fc --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/ByteBufferBackedKeyValueUtil.java @@ -0,0 +1,196 @@ +/* + * 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; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; + +import org.apache.hadoop.hbase.KeyValue.Type; +import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.io.util.StreamUtils; +import org.apache.hadoop.hbase.util.ByteBufferUtils; +import org.apache.hadoop.hbase.util.IterableUtils; +import org.apache.hadoop.io.WritableUtils; + +@InterfaceAudience.Private +public class ByteBufferBackedKeyValueUtil { + public static ByteBufferBackedKeyValue createLastOnRowCol(ByteBufferBackedCell kv) { + return new ByteBufferBackedKeyValue(kv.getRowBuffer(), kv.getRowOffset(), kv.getRowLength(), + kv.getFamilyBuffer(), kv.getFamilyOffset(), kv.getFamilyLength(), kv.getQualifierBuffer(), + kv.getQualifierOffset(), kv.getQualifierLength(), HConstants.OLDEST_TIMESTAMP, + Type.Minimum, null, 0, 0); + } + + public static ByteBufferBackedKeyValue createFirstOnRowColTS(ByteBufferBackedCell kv, long ts) { + return new ByteBufferBackedKeyValue(kv.getRowBuffer(), kv.getRowOffset(), kv.getRowLength(), + kv.getFamilyBuffer(), kv.getFamilyOffset(), kv.getFamilyLength(), kv.getQualifierBuffer(), + kv.getQualifierOffset(), kv.getQualifierLength(), ts, Type.Maximum, kv.getValueBuffer(), + kv.getValueOffset(), kv.getValueLength()); + } + + public static ByteBufferBackedCell createFirstOnRow(final ByteBuffer row, int roffset, short rlength) { + return new ByteBufferBackedKeyValue(row, roffset, rlength, null, 0, 0, null, 0, 0, + HConstants.LATEST_TIMESTAMP, Type.Maximum, null, 0, 0); + } + + public static ByteBufferBackedCell createLastOnRow(final ByteBuffer row) { + return new ByteBufferBackedKeyValue(row, null, null, HConstants.LATEST_TIMESTAMP, Type.Minimum); + } + + public static ByteBufferBackedCell createLastOnRow(final ByteBuffer row, final int roffset, final int rlength, + final ByteBuffer family, final int foffset, final int flength, final ByteBuffer qualifier, + final int qoffset, final int qlength) { + return new ByteBufferBackedKeyValue(row, roffset, rlength, family, foffset, flength, qualifier, qoffset, + qlength, HConstants.OLDEST_TIMESTAMP, Type.Minimum, null, 0, 0); + } + + public static ByteBufferBackedCell createLastOnRow(ByteBufferBackedCell kv) { + return createLastOnRow(kv.getRowBuffer(), kv.getRowOffset(), kv.getRowLength(), null, 0, 0, + null, 0, 0); + } + + public static ByteBufferBackedCell createFirstOnRow(final ByteBuffer row, final long ts) { + return new ByteBufferBackedKeyValue(row, null, null, ts, Type.Maximum); + } + + public static ByteBufferBackedCell createFirstOnRow(final ByteBuffer row) { + return new ByteBufferBackedKeyValue(row, null, null, HConstants.LATEST_TIMESTAMP, Type.Maximum); + } + + public static ByteBufferBackedCell createFirstOnRow(final ByteBuffer row, final ByteBuffer family, + final ByteBuffer qualifier) { + return new ByteBufferBackedKeyValue(row, family, qualifier, HConstants.LATEST_TIMESTAMP, + Type.Maximum); + } + + public static ByteBufferBackedCell createFirstDeleteFamilyOnRow(final ByteBuffer row, + final ByteBuffer family) { + return new ByteBufferBackedKeyValue(row, family, null, HConstants.LATEST_TIMESTAMP, + Type.DeleteFamily); + } + + public static ByteBufferBackedCell createFirstOnRow(final ByteBuffer row, final ByteBuffer f, + final ByteBuffer q, final long ts) { + return new ByteBufferBackedKeyValue(row, f, q, ts, Type.Maximum); + } + + public static ByteBufferBackedCell createFirstOnRow(final ByteBuffer row, final int roffset, + final int rlength, final ByteBuffer family, final int foffset, final int flength, + final ByteBuffer qualifier, final int qoffset, final int qlength) { + return new ByteBufferBackedKeyValue(row, roffset, rlength, family, foffset, flength, qualifier, + qoffset, qlength, HConstants.LATEST_TIMESTAMP, Type.Maximum, null, 0, 0); + } + + public static void oswrite(final ByteBufferBackedCell cell, final OutputStream out, + final boolean withTags) throws IOException { + if (cell instanceof KeyValue) { + KeyValue.oswrite((KeyValue) cell, out, withTags); + } else { + short rlen = cell.getRowLength(); + byte flen = cell.getFamilyLength(); + int qlen = cell.getQualifierLength(); + int vlen = cell.getValueLength(); + int tlen = cell.getTagsLength(); + + // write total length + StreamUtils.writeInt(out, length(rlen, flen, qlen, vlen, tlen, withTags)); + // write key length + StreamUtils.writeInt(out, keyLength(rlen, flen, qlen)); + // write value length + StreamUtils.writeInt(out, vlen); + // Write rowkey - 2 bytes rk length followed by rowkey bytes + StreamUtils.writeShort(out, rlen); + ByteBufferUtils.copyBufferToStream(out, cell.getRowBuffer(), cell.getRowOffset(), rlen); + // Write cf - 1 byte of cf length followed by the family bytes + out.write(flen); + ByteBufferUtils.copyBufferToStream(out, cell.getFamilyBuffer(), cell.getFamilyOffset(), flen); + // write qualifier + ByteBufferUtils.copyBufferToStream(out, cell.getQualifierBuffer(), cell.getQualifierOffset(), + qlen); + // write timestamp + StreamUtils.writeLong(out, cell.getTimestamp()); + // write the type + out.write(cell.getTypeByte()); + // write value + ByteBufferUtils.copyBufferToStream(out, cell.getValueBuffer(), cell.getValueOffset(), vlen); + // write tags if we have to + if (withTags) { + // 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 & (tlen >> 8))); + out.write((byte) (0xff & tlen)); + ByteBufferUtils.copyBufferToStream(out, cell.getTagsBuffer(), cell.getTagsOffset(), tlen); + } + } + } + + /** + * Returns number of bytes this cell would have been used if serialized as in + * {@link KeyValue} + * + * @param cell + * @return the length + */ + public static int length(final Cell cell) { + return length(cell.getRowLength(), cell.getFamilyLength(), cell.getQualifierLength(), + cell.getValueLength(), cell.getTagsLength(), true); + } + + private static int length(short rlen, byte flen, int qlen, int vlen, int tlen, boolean withTags) { + if (withTags) { + return (int) (KeyValue.getKeyValueDataStructureSize(rlen, flen, qlen, vlen, tlen)); + } + return (int) (KeyValue.getKeyValueDataStructureSize(rlen, flen, qlen, vlen)); + } + + /** + * Returns number of bytes this cell's key part would have been used if + * serialized as in {@link KeyValue}. Key includes rowkey, family, qualifier, + * timestamp and type. + * + * @param cell + * @return the key length + */ + public static int keyLength(final Cell cell) { + return keyLength(cell.getRowLength(), cell.getFamilyLength(), cell.getQualifierLength()); + } + + private static int keyLength(short rlen, byte flen, int qlen) { + return (int) KeyValue.getKeyDataStructureSize(rlen, flen, qlen); + } + + public static int lengthWithMvccVersion(final KeyValue kv, final boolean includeMvccVersion) { + int length = kv.getLength(); + if (includeMvccVersion) { + length += WritableUtils.getVIntSize(kv.getMvccVersion()); + } + return length; + } + + public static int totalLengthWithMvccVersion(final Iterable kvs, + final boolean includeMvccVersion) { + int length = 0; + for (KeyValue kv : IterableUtils.nullSafe(kvs)) { + length += lengthWithMvccVersion(kv, includeMvccVersion); + } + return length; + } +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/Cell.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/Cell.java index 8f299cc..da4d2f4 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/Cell.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/Cell.java @@ -195,7 +195,13 @@ public interface Cell { * @return the total length of the tags in the Cell. */ int getTagsLength(); - + + /** + * @return true if the Cell is backed by byte[]. In case of BB backed Cell we + * return false. + */ + boolean hasArray(); + /** * WARNING do not use, expensive. This gets an arraycopy of the cell's value. * diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java index 22b40ec..18acb63 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java @@ -1577,6 +1577,11 @@ public class KeyValue implements Cell, HeapSize, Cloneable, SettableSequenceId, } return tagsLen; } + + @Override + public boolean hasArray() { + return true; + } /** * Returns any tags embedded in the KeyValue. Used in testcases. diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/Tag.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/Tag.java index 644173c..93e5a71 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/Tag.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/Tag.java @@ -19,6 +19,7 @@ */ package org.apache.hadoop.hbase; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; @@ -119,6 +120,11 @@ public class Tag { public byte[] getBuffer() { return this.bytes; } + + public ByteBuffer getAsByteBuffer() { + // FIXME + return ByteBuffer.wrap(this.bytes); + } /** * @return the tag type 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 be8c192..b1b8de2 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 @@ -275,6 +275,11 @@ abstract class BufferedDataBlockEncoder implements DataBlockEncoder { public int getTagsLength() { return tagsLength; } + + @Override + public boolean hasArray() { + return true; + } @Override @Deprecated @@ -474,6 +479,11 @@ abstract class BufferedDataBlockEncoder implements DataBlockEncoder { public int getTagsLength() { return tagsLength; } + + @Override + public boolean hasArray() { + return true; + } @Override @Deprecated diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferUtils.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferUtils.java index 2689946..7aa594a 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferUtils.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ByteBufferUtils.java @@ -22,7 +22,14 @@ import java.io.DataInputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.security.AccessController; +import java.security.PrivilegedAction; + +import sun.misc.Unsafe; +import sun.nio.ch.DirectBuffer; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.classification.InterfaceStability; @@ -135,6 +142,83 @@ public final class ByteBufferUtils { out.write((byte) (value >>> (i * 8))); } } + + public static int putShort(ByteBuffer buffer, int offset, short val) { + if (buffer.capacity() - offset < Bytes.SIZEOF_SHORT) { + throw new IllegalArgumentException("Not enough room to put a short at" + + " offset " + offset + " in a " + buffer.capacity() + " byte array"); + } + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.putShortUnsafe(buffer, offset, val); + } else { + buffer.put(offset + 1, (byte) val); + val >>= 8; + buffer.put(offset, (byte) val); + return offset + Bytes.SIZEOF_SHORT; + } + } + + public static int putShort(ByteBuffer buffer, int offset, int val) { + if (buffer.capacity() - offset < Bytes.SIZEOF_SHORT) { + throw new IllegalArgumentException("Not enough room to put a short at" + + " offset " + offset + " in a " + buffer.capacity() + " byte array"); + } + buffer.put(offset + 1, (byte) val); + val >>= 8; + buffer.put(offset, (byte) val); + return offset + Bytes.SIZEOF_SHORT; + } + + public static int putInt(ByteBuffer buffer, int offset, int val) { + if (buffer.capacity() - offset < Bytes.SIZEOF_INT) { + throw new IllegalArgumentException("Not enough room to put an int at" + + " offset " + offset + " in a " + buffer.capacity() + " byte array"); + } + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.putIntUnsafe(buffer, offset, val); + } else { + for(int i= offset + 3; i > offset; i--) { + buffer.put(i, (byte)val); + val >>>= 8; + } + buffer.put(offset, (byte)val); + return offset + Bytes.SIZEOF_INT; + } + } + + public static int putFloat(ByteBuffer buffer, int offset, float f) { + return putInt(buffer, offset, Float.floatToRawIntBits(f)); + } + + public static int putDouble(ByteBuffer buffer, int offset, double d) { + return putLong(buffer, offset, Double.doubleToLongBits(d)); + } + + public static int putLong(ByteBuffer buffer, int offset, long val) { + if (buffer.capacity() - offset < Bytes.SIZEOF_LONG) { + throw new IllegalArgumentException("Not enough room to put a long at" + + " offset " + offset + " in a " + buffer.capacity() + " byte array"); + } + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.putLongUnsafe(buffer, offset, val); + } else { + for(int i = offset + 7; i > offset; i--) { + buffer.put(i, (byte)val); + val >>>= 8; + } + buffer.put(offset, (byte)val); + return offset + Bytes.SIZEOF_LONG; + } + } + + public static int putByte(ByteBuffer buffer, int offset, byte b) { + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.putByte(buffer, offset, b); + } else { + buffer.put(offset, b); + return offset + 1; + } + } /** * Copy the data to the output stream and update position in buffer. @@ -363,7 +447,7 @@ public final class ByteBufferUtils { * @param destinationOffset * @param length */ - public static void copyFromBufferToBuffer(ByteBuffer out, ByteBuffer in, int sourceOffset, + public static int copyFromBufferToBuffer(ByteBuffer out, ByteBuffer in, int sourceOffset, int destinationOffset, int length) { if (in.hasArray() && out.hasArray()) { System.arraycopy(in.array(), sourceOffset + in.arrayOffset(), out.array(), out.arrayOffset() @@ -373,6 +457,7 @@ public final class ByteBufferUtils { out.put((destinationOffset + i), in.get(sourceOffset + i)); } } + return destinationOffset + length; } /** @@ -492,11 +577,118 @@ public final class ByteBufferUtils { } return output; } + + public static int compareTo(ByteBuffer left, int leftOffset, int leftLength, byte[] right, + int rightOffset, int rightLength) { + // Bring WritableComparator code local + int end1 = leftOffset + leftLength; + int end2 = rightOffset + rightLength; + for (int i = leftOffset - left.arrayOffset(), j = rightOffset; i < end1 + && j < end2; i++, j++) { + int a = (left.get(i) & 0xff); + int b = (right[j] & 0xff); + if (a != b) { + return a - b; + } + } + return leftLength - rightLength; + } + + public static int hashCode(ByteBuffer buffer, int offset, int length) { + int hash = 1; + for (int i = offset; i < offset + length; i++) + hash = (31 * hash) + (int) buffer.get(i); + return hash; + } + + public static int compareTo(byte[] left, int leftOffset, int leftLength, ByteBuffer right, + int rightOffset, int rightLength) { + // Bring WritableComparator code local + int end1 = leftOffset + leftLength; + int end2 = rightOffset + rightLength; + for (int i = leftOffset, j = rightOffset - right.arrayOffset(); i < end1 + && j < end2; i++, j++) { + int a = (left[i] & 0xff); + int b = (right.get(j) & 0xff); + if (a != b) { + return a - b; + } + } + return leftLength - rightLength; + } + + public static boolean equals(final ByteBuffer left, int leftOffset, int leftLen, final ByteBuffer right, + int rightOffset, int rightLen) { + // short circuit case + if (left == right && leftOffset == rightOffset && leftLen == rightLen) { + return true; + } + // different lengths fast check + if (leftLen != rightLen) { + return false; + } + if (leftLen == 0) { + return true; + } + + // Since we're often comparing adjacent sorted data, + // it's usual to have equal arrays except for the very last byte + // so check that first + if (left.get(leftOffset + leftLen - 1 - left.arrayOffset()) != right.get(rightOffset + rightLen + - 1 - right.arrayOffset())) + return false; + + return compareTo(left, leftOffset, leftLen, right, + rightOffset, rightLen) == 0; + } + + public static boolean equals(final ByteBuffer left, int leftOffset, int leftLen, final byte[] right, + int rightOffset, int rightLen) { + // different lengths fast check + if (leftLen != rightLen) { + return false; + } + if (leftLen == 0) { + return true; + } + + // Since we're often comparing adjacent sorted data, + // it's usual to have equal arrays except for the very last byte + // so check that first + if (left.get(leftOffset + leftLen - 1 - left.arrayOffset()) != right[rightOffset + rightLen + - 1]) { + return false; + } + + return compareTo(left, leftOffset, leftLen, right, + rightOffset, rightLen) == 0; + } + + public static boolean equals(final byte[] left, int leftOffset, int leftLen, final ByteBuffer right, + int rightOffset, int rightLen) { + // different lengths fast check + if (leftLen != rightLen) { + return false; + } + if (leftLen == 0) { + return true; + } + + // Since we're often comparing adjacent sorted data, + // it's usual to have equal arrays except for the very last byte + // so check that first + if (left[leftOffset + leftLen - 1] != right.get(rightOffset + rightLen + - 1- right.arrayOffset())) { + return false; + } + + return compareTo(left, leftOffset, leftLen, right, + rightOffset, rightLen) == 0; + } public static int compareTo(ByteBuffer buf1, int o1, int len1, ByteBuffer buf2, int o2, int len2) { - if (buf1.hasArray() && buf2.hasArray()) { - return Bytes.compareTo(buf1.array(), buf1.arrayOffset() + o1, len1, buf2.array(), - buf2.arrayOffset() + o2, len2); + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.compareTo(buf1, o1, len1, buf2, o2, len2); } int end1 = o1 + len1; int end2 = o2 + len2; @@ -509,4 +701,317 @@ public final class ByteBufferUtils { } return len1 - len2; } + + public static int readAsInt(ByteBuffer buffer, int offset) { + return readAsInt(buffer, offset, Bytes.SIZEOF_INT); + } + + public static int readAsInt(ByteBuffer buffer, int offset, final int length) { + if (offset + length > buffer.capacity()) { + throw new IllegalArgumentException("offset (" + offset + ") + length (" + length + + ") exceed the" + " capacity of the array: " + buffer.capacity()); + } + if(UnsafeUtil.isAvailable()) { + return UnsafeUtil.toIntUnsafe(buffer, offset); + } else { + int n = 0; + for (int i = offset; i < (offset + length); i++) { + n <<= 8; + n ^= buffer.get(i) & 0xFF; + } + return n; + } + } + + public static String toStringBinary(final ByteBuffer b, int off, int len) { + StringBuilder result = new StringBuilder(); + // Just in case we are passed a 'len' that is > buffer length... + if (off >= b.capacity()) return result.toString(); + if (off + len > b.capacity()) len = b.capacity() - off; + for (int i = off; i < off + len ; ++i ) { + int ch = b.get(i) & 0xFF; + if ( (ch >= '0' && ch <= '9') + || (ch >= 'A' && ch <= 'Z') + || (ch >= 'a' && ch <= 'z') + || " `~!@#$%^&*()-_=+[]{}|;:'\",.<>/?".indexOf(ch) >= 0 ) { + result.append((char)ch); + } else { + result.append(String.format("\\x%02X", ch)); + } + } + return result.toString(); + } + + public static byte readAsByte(ByteBuffer buffer, int offset) { + if (offset + Bytes.SIZEOF_BYTE > buffer.capacity()) { + throw new IllegalArgumentException("offset (" + offset + ") + length (" + Bytes.SIZEOF_BYTE + + ") exceed the" + " capacity of the array: " + buffer.capacity()); + } + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.get(buffer, offset); + } else { + return buffer.get(offset); + } + } + + public static long readAsLong(ByteBuffer buffer, int offset) { + return readAsLong(buffer, offset, Bytes.SIZEOF_LONG); + } + public static long readAsLong(ByteBuffer buffer, int offset, final int length) { + if (offset + length > buffer.capacity()) { + throw new IllegalArgumentException("offset (" + offset + ") + length (" + length + + ") exceed the" + " capacity of the array: " + buffer.capacity()); + } + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.toLongUnsafe(buffer, offset); + } else { + long l = 0; + for(int i = offset; i < offset + length; i++) { + l <<= 8; + l ^= buffer.get(i) & 0xFF; + } + return l; + } + } + public static short readAsShort(ByteBuffer buffer, int offset) { + return readAsShort(buffer, offset, Bytes.SIZEOF_SHORT); + } + + public static short readAsShort(ByteBuffer buffer, int offset, final int length) { + if (offset + length > buffer.capacity()) { + throw new IllegalArgumentException("offset (" + offset + ") + length (" + length + + ") exceed the" + " capacity of the array: " + buffer.capacity()); + } + if (UnsafeUtil.isAvailable()) { + return UnsafeUtil.toShortUnsafe(buffer, offset); + } else { + short n = 0; + n ^= buffer.get(offset) & 0xFF; + n <<= 8; + n ^= buffer.get(offset + 1) & 0xFF; + return n; + } + } + + static class UnsafeUtil { + static final Unsafe theUnsafe; + + /** The offset to the first element in a byte array. */ + static long BYTE_ARRAY_BASE_OFFSET; + + static { + theUnsafe = (Unsafe) AccessController.doPrivileged(new PrivilegedAction() { + @Override + public Object run() { + try { + Field f = Unsafe.class.getDeclaredField("theUnsafe"); + f.setAccessible(true); + return f.get(null); + } catch (Exception e) { + return null; + } + } + }); + if (theUnsafe != null) + BYTE_ARRAY_BASE_OFFSET = theUnsafe.arrayBaseOffset(byte[].class); + } + + static final boolean littleEndian = ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN); + + static boolean isAvailable() { + return theUnsafe != null; + } + + static boolean lessThanUnsignedLong(long x1, long x2) { + if (littleEndian) { + x1 = Long.reverseBytes(x1); + x2 = Long.reverseBytes(x2); + } + return (x1 + Long.MIN_VALUE) < (x2 + Long.MIN_VALUE); + } + + static boolean lessThanUnsignedInt(int x1, int x2) { + if (littleEndian) { + x1 = Integer.reverseBytes(x1); + x2 = Integer.reverseBytes(x2); + } + return (x1 & 0xffffffffL) < (x2 & 0xffffffffL); + } + + static boolean lessThanUnsignedShort(short x1, short x2) { + if (littleEndian) { + x1 = Short.reverseBytes(x1); + x2 = Short.reverseBytes(x2); + } + return (x1 & 0xffff) < (x2 & 0xffff); + } + + static long getLong(ByteBuffer buf, int offset) { + if (buf.isDirect()) { + return theUnsafe.getLong(((DirectBuffer) buf).address() + offset); + } else { + assert buf.hasArray(); + return theUnsafe.getLong(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset); + } + } + + static int putShort(ByteBuffer buf, int offset, short val) { + if (buf.isDirect()) { + theUnsafe.putShort(buf, ((DirectBuffer) buf).address() + offset, val); + } else { + assert buf.hasArray(); + theUnsafe.putShort(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset, val); + } + return offset + Bytes.SIZEOF_SHORT; + } + + static int putInt(ByteBuffer buf, int offset, int val) { + if (buf.isDirect()) { + theUnsafe.putInt(buf, ((DirectBuffer) buf).address() + offset, val); + } else { + assert buf.hasArray(); + theUnsafe.putInt(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset, val); + } + return offset + Bytes.SIZEOF_INT; + } + + static int putLong(ByteBuffer buf, int offset, long val) { + if (buf.isDirect()) { + theUnsafe.putLong(buf, ((DirectBuffer) buf).address() + offset, val); + } else { + assert buf.hasArray(); + theUnsafe.putLong(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset, val); + } + return offset + Bytes.SIZEOF_LONG; + } + + static int getInt(ByteBuffer buf, int offset) { + if (buf.isDirect()) { + return theUnsafe.getInt(((DirectBuffer) buf).address() + offset); + } else { + assert buf.hasArray(); + return theUnsafe.getInt(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset); + } + } + + static byte get(ByteBuffer buf, int offset) { + if (buf.isDirect()) { + return theUnsafe.getByte(((DirectBuffer) buf).address() + offset); + } else { + assert buf.hasArray(); + return theUnsafe.getByte(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset); + } + } + + static int putByte(ByteBuffer buf, int offset, byte b) { + if (buf.isDirect()) { + UnsafeUtil.theUnsafe.putByte(((DirectBuffer) buf).address() + offset, b); + } else { + assert buf.hasArray(); + UnsafeUtil.theUnsafe.putByte(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + + offset, b); + } + return offset + 1; + } + + static short getShort(ByteBuffer buf, int offset) { + if (buf.isDirect()) { + return theUnsafe.getShort(((DirectBuffer) buf).address() + offset); + } else { + assert buf.hasArray(); + return theUnsafe.getShort(buf.array(), BYTE_ARRAY_BASE_OFFSET + buf.arrayOffset() + offset); + } + } + + static int toIntUnsafe(ByteBuffer buf, int offset) { + if (UnsafeUtil.littleEndian) { + return Integer.reverseBytes(getInt(buf, offset)); + } else { + return getInt(buf, offset); + } + } + + static short toShortUnsafe(ByteBuffer buf, int offset) { + if (UnsafeUtil.littleEndian) { + return Short.reverseBytes(getShort(buf, offset)); + } else { + return getShort(buf, offset); + } + } + + static long toLongUnsafe(ByteBuffer buf, int offset) { + if (UnsafeUtil.littleEndian) { + return Long.reverseBytes(getLong(buf, offset)); + } else { + return getLong(buf, offset); + } + } + + static int putShortUnsafe(ByteBuffer buffer, int offset, short val) { + if (UnsafeUtil.littleEndian) { + val = Short.reverseBytes(val); + } + return putShort(buffer, offset, val); + } + + static int putIntUnsafe(ByteBuffer buffer, int offset, int val) { + if (UnsafeUtil.littleEndian) { + val = Integer.reverseBytes(val); + } + return putInt(buffer, offset, val); + } + + static int putLongUnsafe(ByteBuffer buffer, int offset, long val) { + if (UnsafeUtil.littleEndian) { + val = Long.reverseBytes(val); + } + return putLong(buffer, offset, val); + } + + static int compareTo(ByteBuffer buffer1, int offset1, int length1, ByteBuffer buffer2, + int offset2, int length2) { + final int minLength = Math.min(length1, length2); + final int minWords = minLength / Bytes.SIZEOF_LONG; + + /* + * Compare 8 bytes at a time. Benchmarking shows comparing 8 bytes at a + * time is no slower than comparing 4 bytes at a time even on 32-bit. On + * the other hand, it is substantially faster on 64-bit. + */ + for (int i = 0; i < minWords * Bytes.SIZEOF_LONG; i += Bytes.SIZEOF_LONG) { + long lw = getLong(buffer1, offset1 + i); + long rw = getLong(buffer2, offset2 + i); + long diff = lw ^ rw; + if (diff != 0) { + return lessThanUnsignedLong(lw, rw) ? -1 : 1; + } + } + int offset = minWords * Bytes.SIZEOF_LONG; + + if (minLength - offset >= Bytes.SIZEOF_INT) { + int il = getInt(buffer1, offset1 + offset); + int ir = getInt(buffer2, offset2 + offset); + if (il != ir) { + return lessThanUnsignedInt(il, ir) ? -1 : 1; + } + offset += Bytes.SIZEOF_INT; + } + if (minLength - offset >= Bytes.SIZEOF_SHORT) { + short sl = getShort(buffer1, offset1 + offset); + short sr = getShort(buffer2, offset2 + offset); + if (sl != sr) { + return lessThanUnsignedShort(sl, sr) ? -1 : 1; + } + offset += Bytes.SIZEOF_SHORT; + } + if (minLength - offset == 1) { + int a = (buffer1.get(offset1 + offset) & 0xff); + int b = (buffer2.get(offset2 + offset) & 0xff); + if (a != b) { + return a - b; + } + } + return length1 - length2; + } + } } diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/TestCellUtil.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/TestCellUtil.java index 5c18b51..20ef1e1 100644 --- a/hbase-common/src/test/java/org/apache/hadoop/hbase/TestCellUtil.java +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/TestCellUtil.java @@ -18,7 +18,7 @@ package org.apache.hadoop.hbase; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; import java.io.IOException; import java.util.ArrayList; @@ -222,6 +222,12 @@ public class TestCellUtil { // TODO Auto-generated method stub return 0; } + + @Override + public boolean hasArray() { + // TODO Auto-generated method stub + return true; + } }; /** diff --git a/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/PrefixTreeSeeker.java b/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/PrefixTreeSeeker.java index 2db0f29..3542c4a 100644 --- a/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/PrefixTreeSeeker.java +++ b/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/PrefixTreeSeeker.java @@ -442,6 +442,11 @@ public class PrefixTreeSeeker implements EncodedSeeker { public byte[] getRow() { return this.row; } + + @Override + public boolean hasArray() { + return true; + } @Override public String toString() { diff --git a/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/decode/PrefixTreeCell.java b/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/decode/PrefixTreeCell.java index 97eed62..34a5e66 100644 --- a/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/decode/PrefixTreeCell.java +++ b/hbase-prefix-tree/src/main/java/org/apache/hadoop/hbase/codec/prefixtree/decode/PrefixTreeCell.java @@ -234,6 +234,11 @@ public class PrefixTreeCell implements Cell, SettableSequenceId, Comparable