You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ec...@apache.org on 2013/02/25 23:50:29 UTC
svn commit: r1449950 [29/35] - in /hbase/trunk: ./ hbase-client/
hbase-client/src/ hbase-client/src/main/ hbase-client/src/main/java/
hbase-client/src/main/java/org/ hbase-client/src/main/java/org/apache/
hbase-client/src/main/java/org/apache/hadoop/ h...
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,372 @@
+/*
+ * 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 java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.SamePrefixComparator;
+import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
+import org.apache.hadoop.hbase.io.hfile.BlockType;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.WritableUtils;
+
+/**
+ * Base class for all data block encoders that use a buffer.
+ */
+@InterfaceAudience.Private
+abstract class BufferedDataBlockEncoder implements DataBlockEncoder {
+
+ private static int INITIAL_KEY_BUFFER_SIZE = 512;
+
+ @Override
+ public ByteBuffer decodeKeyValues(DataInputStream source,
+ boolean includesMemstoreTS) throws IOException {
+ return decodeKeyValues(source, 0, 0, includesMemstoreTS);
+ }
+
+ protected static class SeekerState {
+ protected int valueOffset = -1;
+ protected int keyLength;
+ protected int valueLength;
+ protected int lastCommonPrefix;
+
+ /** We need to store a copy of the key. */
+ protected byte[] keyBuffer = new byte[INITIAL_KEY_BUFFER_SIZE];
+
+ protected long memstoreTS;
+ protected int nextKvOffset;
+
+ protected boolean isValid() {
+ return valueOffset != -1;
+ }
+
+ protected void invalidate() {
+ valueOffset = -1;
+ }
+
+ protected void ensureSpaceForKey() {
+ if (keyLength > keyBuffer.length) {
+ // rare case, but we need to handle arbitrary length of key
+ int newKeyBufferLength = Math.max(keyBuffer.length, 1) * 2;
+ while (keyLength > newKeyBufferLength) {
+ newKeyBufferLength *= 2;
+ }
+ byte[] newKeyBuffer = new byte[newKeyBufferLength];
+ System.arraycopy(keyBuffer, 0, newKeyBuffer, 0, keyBuffer.length);
+ keyBuffer = newKeyBuffer;
+ }
+ }
+
+ /**
+ * Copy the state from the next one into this instance (the previous state
+ * placeholder). Used to save the previous state when we are advancing the
+ * seeker to the next key/value.
+ */
+ protected void copyFromNext(SeekerState nextState) {
+ if (keyBuffer.length != nextState.keyBuffer.length) {
+ keyBuffer = nextState.keyBuffer.clone();
+ } else if (!isValid()) {
+ // Note: we can only call isValid before we override our state, so this
+ // comes before all the assignments at the end of this method.
+ System.arraycopy(nextState.keyBuffer, 0, keyBuffer, 0,
+ nextState.keyLength);
+ } else {
+ // don't copy the common prefix between this key and the previous one
+ System.arraycopy(nextState.keyBuffer, nextState.lastCommonPrefix,
+ keyBuffer, nextState.lastCommonPrefix, nextState.keyLength
+ - nextState.lastCommonPrefix);
+ }
+
+ valueOffset = nextState.valueOffset;
+ keyLength = nextState.keyLength;
+ valueLength = nextState.valueLength;
+ lastCommonPrefix = nextState.lastCommonPrefix;
+ nextKvOffset = nextState.nextKvOffset;
+ memstoreTS = nextState.memstoreTS;
+ }
+
+ }
+
+ protected abstract static class
+ BufferedEncodedSeeker<STATE extends SeekerState>
+ implements EncodedSeeker {
+
+ protected final RawComparator<byte[]> comparator;
+ protected final SamePrefixComparator<byte[]> samePrefixComparator;
+ protected ByteBuffer currentBuffer;
+ protected STATE current = createSeekerState(); // always valid
+ protected STATE previous = createSeekerState(); // may not be valid
+
+ @SuppressWarnings("unchecked")
+ public BufferedEncodedSeeker(RawComparator<byte[]> comparator) {
+ this.comparator = comparator;
+ if (comparator instanceof SamePrefixComparator) {
+ this.samePrefixComparator = (SamePrefixComparator<byte[]>) comparator;
+ } else {
+ this.samePrefixComparator = null;
+ }
+ }
+
+ @Override
+ public void setCurrentBuffer(ByteBuffer buffer) {
+ currentBuffer = buffer;
+ decodeFirst();
+ previous.invalidate();
+ }
+
+ @Override
+ public ByteBuffer getKeyDeepCopy() {
+ ByteBuffer keyBuffer = ByteBuffer.allocate(current.keyLength);
+ keyBuffer.put(current.keyBuffer, 0, current.keyLength);
+ return keyBuffer;
+ }
+
+ @Override
+ public ByteBuffer getValueShallowCopy() {
+ return ByteBuffer.wrap(currentBuffer.array(),
+ currentBuffer.arrayOffset() + current.valueOffset,
+ current.valueLength);
+ }
+
+ @Override
+ public ByteBuffer getKeyValueBuffer() {
+ ByteBuffer kvBuffer = ByteBuffer.allocate(
+ 2 * Bytes.SIZEOF_INT + current.keyLength + current.valueLength);
+ kvBuffer.putInt(current.keyLength);
+ kvBuffer.putInt(current.valueLength);
+ kvBuffer.put(current.keyBuffer, 0, current.keyLength);
+ kvBuffer.put(currentBuffer.array(),
+ currentBuffer.arrayOffset() + current.valueOffset,
+ current.valueLength);
+ return kvBuffer;
+ }
+
+ @Override
+ public KeyValue getKeyValue() {
+ ByteBuffer kvBuf = getKeyValueBuffer();
+ KeyValue kv = new KeyValue(kvBuf.array(), kvBuf.arrayOffset());
+ kv.setMemstoreTS(current.memstoreTS);
+ return kv;
+ }
+
+ @Override
+ public void rewind() {
+ currentBuffer.rewind();
+ decodeFirst();
+ previous.invalidate();
+ }
+
+ @Override
+ public boolean next() {
+ if (!currentBuffer.hasRemaining()) {
+ return false;
+ }
+ decodeNext();
+ previous.invalidate();
+ return true;
+ }
+
+ @Override
+ public int seekToKeyInBlock(byte[] key, int offset, int length,
+ boolean seekBefore) {
+ int commonPrefix = 0;
+ previous.invalidate();
+ do {
+ int comp;
+ if (samePrefixComparator != null) {
+ commonPrefix = Math.min(commonPrefix, current.lastCommonPrefix);
+
+ // extend commonPrefix
+ commonPrefix += ByteBufferUtils.findCommonPrefix(
+ key, offset + commonPrefix, length - commonPrefix,
+ current.keyBuffer, commonPrefix,
+ current.keyLength - commonPrefix);
+
+ comp = samePrefixComparator.compareIgnoringPrefix(commonPrefix, key,
+ offset, length, current.keyBuffer, 0, current.keyLength);
+ } else {
+ comp = comparator.compare(key, offset, length,
+ current.keyBuffer, 0, current.keyLength);
+ }
+
+ if (comp == 0) { // exact match
+ if (seekBefore) {
+ if (!previous.isValid()) {
+ // The caller (seekBefore) has to ensure that we are not at the
+ // first key in the block.
+ throw new IllegalStateException("Cannot seekBefore if " +
+ "positioned at the first key in the block: key=" +
+ Bytes.toStringBinary(key, offset, length));
+ }
+ moveToPrevious();
+ return 1;
+ }
+ return 0;
+ }
+
+ if (comp < 0) { // already too large, check previous
+ if (previous.isValid()) {
+ moveToPrevious();
+ }
+ return 1;
+ }
+
+ // move to next, if more data is available
+ if (currentBuffer.hasRemaining()) {
+ previous.copyFromNext(current);
+ decodeNext();
+ } else {
+ break;
+ }
+ } while (true);
+
+ // we hit the end of the block, not an exact match
+ return 1;
+ }
+
+ private void moveToPrevious() {
+ if (!previous.isValid()) {
+ throw new IllegalStateException(
+ "Can move back only once and not in first key in the block.");
+ }
+
+ STATE tmp = previous;
+ previous = current;
+ current = tmp;
+
+ // move after last key value
+ currentBuffer.position(current.nextKvOffset);
+
+ previous.invalidate();
+ }
+
+ @SuppressWarnings("unchecked")
+ protected STATE createSeekerState() {
+ // This will fail for non-default seeker state if the subclass does not
+ // override this method.
+ return (STATE) new SeekerState();
+ }
+
+ abstract protected void decodeFirst();
+ abstract protected void decodeNext();
+ }
+
+ protected final void afterEncodingKeyValue(ByteBuffer in,
+ DataOutputStream out, boolean includesMemstoreTS) {
+ if (includesMemstoreTS) {
+ // Copy memstore timestamp from the byte buffer to the output stream.
+ long memstoreTS = -1;
+ try {
+ memstoreTS = ByteBufferUtils.readVLong(in);
+ WritableUtils.writeVLong(out, memstoreTS);
+ } catch (IOException ex) {
+ throw new RuntimeException("Unable to copy memstore timestamp " +
+ memstoreTS + " after encoding a key/value");
+ }
+ }
+ }
+
+ protected final void afterDecodingKeyValue(DataInputStream source,
+ ByteBuffer dest, boolean includesMemstoreTS) {
+ if (includesMemstoreTS) {
+ long memstoreTS = -1;
+ try {
+ // Copy memstore timestamp from the data input stream to the byte
+ // buffer.
+ memstoreTS = WritableUtils.readVLong(source);
+ ByteBufferUtils.writeVLong(dest, memstoreTS);
+ } catch (IOException ex) {
+ throw new RuntimeException("Unable to copy memstore timestamp " +
+ memstoreTS + " after decoding a key/value");
+ }
+ }
+ }
+
+ @Override
+ public HFileBlockEncodingContext newDataBlockEncodingContext(
+ Algorithm compressionAlgorithm,
+ DataBlockEncoding encoding, byte[] header) {
+ return new HFileBlockDefaultEncodingContext(
+ compressionAlgorithm, encoding, header);
+ }
+
+ @Override
+ public HFileBlockDecodingContext newDataBlockDecodingContext(
+ Algorithm compressionAlgorithm) {
+ return new HFileBlockDefaultDecodingContext(compressionAlgorithm);
+ }
+
+ /**
+ * Compress KeyValues and write them to output buffer.
+ * @param out Where to write compressed data.
+ * @param in Source of KeyValue for compression.
+ * @param includesMemstoreTS true if including memstore timestamp after every
+ * key-value pair
+ * @throws IOException If there is an error writing to output stream.
+ */
+ public abstract void internalEncodeKeyValues(DataOutputStream out,
+ ByteBuffer in, boolean includesMemstoreTS) throws IOException;
+
+ @Override
+ public void encodeKeyValues(ByteBuffer in,
+ boolean includesMemstoreTS,
+ HFileBlockEncodingContext blkEncodingCtx) throws IOException {
+ if (blkEncodingCtx.getClass() != HFileBlockDefaultEncodingContext.class) {
+ throw new IOException (this.getClass().getName() + " only accepts "
+ + HFileBlockDefaultEncodingContext.class.getName() + " as the " +
+ "encoding context.");
+ }
+
+ HFileBlockDefaultEncodingContext encodingCtx =
+ (HFileBlockDefaultEncodingContext) blkEncodingCtx;
+ encodingCtx.prepareEncoding();
+ DataOutputStream dataOut =
+ ((HFileBlockDefaultEncodingContext) encodingCtx)
+ .getOutputStreamForEncoder();
+ internalEncodeKeyValues(dataOut, in, includesMemstoreTS);
+ if (encodingCtx.getDataBlockEncoding() != DataBlockEncoding.NONE) {
+ encodingCtx.postEncoding(BlockType.ENCODED_DATA);
+ } else {
+ encodingCtx.postEncoding(BlockType.DATA);
+ }
+ }
+
+ /**
+ * Asserts that there is at least the given amount of unfilled space
+ * remaining in the given buffer.
+ * @param out typically, the buffer we are writing to
+ * @param length the required space in the buffer
+ * @throws EncoderBufferTooSmallException If there are no enough bytes.
+ */
+ protected static void ensureSpace(ByteBuffer out, int length)
+ throws EncoderBufferTooSmallException {
+ if (out.position() + length > out.limit()) {
+ throw new EncoderBufferTooSmallException(
+ "Buffer position=" + out.position() +
+ ", buffer limit=" + out.limit() +
+ ", length to be written=" + length);
+ }
+ }
+
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CompressionState.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CompressionState.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CompressionState.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CompressionState.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,118 @@
+/*
+ * 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 java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+
+/**
+ * Stores the state of data block encoder at the beginning of new key.
+ */
+@InterfaceAudience.Private
+class CompressionState {
+ int keyLength;
+ int valueLength;
+
+ short rowLength;
+ int prevOffset = FIRST_KEY;
+ byte familyLength;
+ int qualifierLength;
+ byte type;
+
+ private final static int FIRST_KEY = -1;
+
+ boolean isFirst() {
+ return prevOffset == FIRST_KEY;
+ }
+
+ /**
+ * Analyze the key and fill the state.
+ * Uses mark() and reset() in ByteBuffer.
+ * @param in Buffer at the position where key starts
+ * @param keyLength Length of key in bytes
+ * @param valueLength Length of values in bytes
+ */
+ void readKey(ByteBuffer in, int keyLength, int valueLength) {
+ readKey(in, keyLength, valueLength, 0, null);
+ }
+
+ /**
+ * Analyze the key and fill the state assuming we know previous state.
+ * Uses mark() and reset() in ByteBuffer to avoid moving the position.
+ * <p>
+ * This method overrides all the fields of this instance, except
+ * {@link #prevOffset}, which is usually manipulated directly by encoders
+ * and decoders.
+ * @param in Buffer at the position where key starts
+ * @param keyLength Length of key in bytes
+ * @param valueLength Length of values in bytes
+ * @param commonPrefix how many first bytes are common with previous KeyValue
+ * @param previousState State from previous KeyValue
+ */
+ void readKey(ByteBuffer in, int keyLength, int valueLength,
+ int commonPrefix, CompressionState previousState) {
+ this.keyLength = keyLength;
+ this.valueLength = valueLength;
+
+ // fill the state
+ in.mark(); // mark beginning of key
+
+ if (commonPrefix < KeyValue.ROW_LENGTH_SIZE) {
+ rowLength = in.getShort();
+ ByteBufferUtils.skip(in, rowLength);
+
+ familyLength = in.get();
+
+ qualifierLength = keyLength - rowLength - familyLength -
+ KeyValue.KEY_INFRASTRUCTURE_SIZE;
+ ByteBufferUtils.skip(in, familyLength + qualifierLength);
+ } else {
+ rowLength = previousState.rowLength;
+ familyLength = previousState.familyLength;
+ qualifierLength = previousState.qualifierLength +
+ keyLength - previousState.keyLength;
+ ByteBufferUtils.skip(in, (KeyValue.ROW_LENGTH_SIZE +
+ KeyValue.FAMILY_LENGTH_SIZE) +
+ rowLength + familyLength + qualifierLength);
+ }
+
+ readTimestamp(in);
+
+ type = in.get();
+
+ in.reset();
+ }
+
+ protected void readTimestamp(ByteBuffer in) {
+ // used in subclasses to add timestamp to state
+ ByteBufferUtils.skip(in, KeyValue.TIMESTAMP_SIZE);
+ }
+
+ void copyFrom(CompressionState state) {
+ keyLength = state.keyLength;
+ valueLength = state.valueLength;
+
+ rowLength = state.rowLength;
+ prevOffset = state.prevOffset;
+ familyLength = state.familyLength;
+ qualifierLength = state.qualifierLength;
+ type = state.type;
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CopyKeyDataBlockEncoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CopyKeyDataBlockEncoder.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CopyKeyDataBlockEncoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/CopyKeyDataBlockEncoder.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,98 @@
+/*
+ * 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 java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.RawComparator;
+
+/**
+ * Just copy data, do not do any kind of compression. Use for comparison and
+ * benchmarking.
+ */
+@InterfaceAudience.Private
+public class CopyKeyDataBlockEncoder extends BufferedDataBlockEncoder {
+ @Override
+ public void internalEncodeKeyValues(DataOutputStream out,
+ ByteBuffer in, boolean includesMemstoreTS) throws IOException {
+ in.rewind();
+ ByteBufferUtils.putInt(out, in.limit());
+ ByteBufferUtils.moveBufferToStream(out, in, in.limit());
+ }
+
+ @Override
+ public ByteBuffer decodeKeyValues(DataInputStream source,
+ int preserveHeaderLength, int skipLastBytes, boolean includesMemstoreTS)
+ throws IOException {
+ int decompressedSize = source.readInt();
+ ByteBuffer buffer = ByteBuffer.allocate(decompressedSize +
+ preserveHeaderLength);
+ buffer.position(preserveHeaderLength);
+ ByteBufferUtils.copyFromStreamToBuffer(buffer, source, decompressedSize);
+
+ return buffer;
+ }
+
+ @Override
+ public ByteBuffer getFirstKeyInBlock(ByteBuffer block) {
+ int keyLength = block.getInt(Bytes.SIZEOF_INT);
+ return ByteBuffer.wrap(block.array(),
+ block.arrayOffset() + 3 * Bytes.SIZEOF_INT, keyLength).slice();
+ }
+
+
+ @Override
+ public String toString() {
+ return CopyKeyDataBlockEncoder.class.getSimpleName();
+ }
+
+ @Override
+ public EncodedSeeker createSeeker(RawComparator<byte[]> comparator,
+ final boolean includesMemstoreTS) {
+ return new BufferedEncodedSeeker<SeekerState>(comparator) {
+ @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 (includesMemstoreTS) {
+ 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();
+ }
+ };
+ }
+
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,550 @@
+/*
+ * 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 java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.RawComparator;
+
+/**
+ * Compress using:
+ * - store size of common prefix
+ * - save column family once, it is same within HFile
+ * - use integer compression for key, value and prefix (7-bit encoding)
+ * - use bits to avoid duplication key length, value length
+ * and type if it same as previous
+ * - store in 3 bits length of timestamp field
+ * - allow diff in timestamp instead of actual value
+ *
+ * Format:
+ * - 1 byte: flag
+ * - 1-5 bytes: key length (only if FLAG_SAME_KEY_LENGTH is not set in flag)
+ * - 1-5 bytes: value length (only if FLAG_SAME_VALUE_LENGTH is not set in flag)
+ * - 1-5 bytes: prefix length
+ * - ... bytes: rest of the row (if prefix length is small enough)
+ * - ... bytes: qualifier (or suffix depending on prefix length)
+ * - 1-8 bytes: timestamp or diff
+ * - 1 byte: type (only if FLAG_SAME_TYPE is not set in the flag)
+ * - ... bytes: value
+ */
+@InterfaceAudience.Private
+public class DiffKeyDeltaEncoder extends BufferedDataBlockEncoder {
+ static final int FLAG_SAME_KEY_LENGTH = 1;
+ static final int FLAG_SAME_VALUE_LENGTH = 1 << 1;
+ static final int FLAG_SAME_TYPE = 1 << 2;
+ static final int FLAG_TIMESTAMP_IS_DIFF = 1 << 3;
+ static final int MASK_TIMESTAMP_LENGTH = (1 << 4) | (1 << 5) | (1 << 6);
+ static final int SHIFT_TIMESTAMP_LENGTH = 4;
+ static final int FLAG_TIMESTAMP_SIGN = 1 << 7;
+
+ protected static class DiffCompressionState extends CompressionState {
+ long timestamp;
+ byte[] familyNameWithSize;
+
+ @Override
+ protected void readTimestamp(ByteBuffer in) {
+ timestamp = in.getLong();
+ }
+
+ @Override
+ void copyFrom(CompressionState state) {
+ super.copyFrom(state);
+ DiffCompressionState state2 = (DiffCompressionState) state;
+ timestamp = state2.timestamp;
+ }
+ }
+
+ private void compressSingleKeyValue(DiffCompressionState previousState,
+ DiffCompressionState currentState, DataOutputStream out,
+ ByteBuffer in) throws IOException {
+ byte flag = 0;
+ int kvPos = in.position();
+ int keyLength = in.getInt();
+ int valueLength = in.getInt();
+
+ long timestamp;
+ long diffTimestamp = 0;
+ int diffTimestampFitsInBytes = 0;
+
+ int commonPrefix;
+
+ int timestampFitsInBytes;
+
+ if (previousState.isFirst()) {
+ currentState.readKey(in, keyLength, valueLength);
+ currentState.prevOffset = kvPos;
+ timestamp = currentState.timestamp;
+ if (timestamp < 0) {
+ flag |= FLAG_TIMESTAMP_SIGN;
+ timestamp = -timestamp;
+ }
+ timestampFitsInBytes = ByteBufferUtils.longFitsIn(timestamp);
+
+ flag |= (timestampFitsInBytes - 1) << SHIFT_TIMESTAMP_LENGTH;
+ commonPrefix = 0;
+
+ // put column family
+ in.mark();
+ ByteBufferUtils.skip(in, currentState.rowLength
+ + KeyValue.ROW_LENGTH_SIZE);
+ ByteBufferUtils.moveBufferToStream(out, in, currentState.familyLength
+ + KeyValue.FAMILY_LENGTH_SIZE);
+ in.reset();
+ } else {
+ // find a common prefix and skip it
+ commonPrefix =
+ ByteBufferUtils.findCommonPrefix(in, in.position(),
+ previousState.prevOffset + KeyValue.ROW_OFFSET, keyLength
+ - KeyValue.TIMESTAMP_TYPE_SIZE);
+ // don't compress timestamp and type using prefix
+
+ currentState.readKey(in, keyLength, valueLength,
+ commonPrefix, previousState);
+ currentState.prevOffset = kvPos;
+ timestamp = currentState.timestamp;
+ boolean negativeTimestamp = timestamp < 0;
+ if (negativeTimestamp) {
+ timestamp = -timestamp;
+ }
+ timestampFitsInBytes = ByteBufferUtils.longFitsIn(timestamp);
+
+ if (keyLength == previousState.keyLength) {
+ flag |= FLAG_SAME_KEY_LENGTH;
+ }
+ if (valueLength == previousState.valueLength) {
+ flag |= FLAG_SAME_VALUE_LENGTH;
+ }
+ if (currentState.type == previousState.type) {
+ flag |= FLAG_SAME_TYPE;
+ }
+
+ // encode timestamp
+ diffTimestamp = previousState.timestamp - currentState.timestamp;
+ boolean minusDiffTimestamp = diffTimestamp < 0;
+ if (minusDiffTimestamp) {
+ diffTimestamp = -diffTimestamp;
+ }
+ diffTimestampFitsInBytes = ByteBufferUtils.longFitsIn(diffTimestamp);
+ if (diffTimestampFitsInBytes < timestampFitsInBytes) {
+ flag |= (diffTimestampFitsInBytes - 1) << SHIFT_TIMESTAMP_LENGTH;
+ flag |= FLAG_TIMESTAMP_IS_DIFF;
+ if (minusDiffTimestamp) {
+ flag |= FLAG_TIMESTAMP_SIGN;
+ }
+ } else {
+ flag |= (timestampFitsInBytes - 1) << SHIFT_TIMESTAMP_LENGTH;
+ if (negativeTimestamp) {
+ flag |= FLAG_TIMESTAMP_SIGN;
+ }
+ }
+ }
+
+ out.write(flag);
+
+ if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ ByteBufferUtils.putCompressedInt(out, keyLength);
+ }
+ if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
+ ByteBufferUtils.putCompressedInt(out, valueLength);
+ }
+
+ ByteBufferUtils.putCompressedInt(out, commonPrefix);
+ ByteBufferUtils.skip(in, commonPrefix);
+
+ if (previousState.isFirst() ||
+ commonPrefix < currentState.rowLength + KeyValue.ROW_LENGTH_SIZE) {
+ int restRowLength =
+ currentState.rowLength + KeyValue.ROW_LENGTH_SIZE - commonPrefix;
+ ByteBufferUtils.moveBufferToStream(out, in, restRowLength);
+ ByteBufferUtils.skip(in, currentState.familyLength +
+ KeyValue.FAMILY_LENGTH_SIZE);
+ ByteBufferUtils.moveBufferToStream(out, in, currentState.qualifierLength);
+ } else {
+ ByteBufferUtils.moveBufferToStream(out, in,
+ keyLength - commonPrefix - KeyValue.TIMESTAMP_TYPE_SIZE);
+ }
+
+ if ((flag & FLAG_TIMESTAMP_IS_DIFF) == 0) {
+ ByteBufferUtils.putLong(out, timestamp, timestampFitsInBytes);
+ } else {
+ ByteBufferUtils.putLong(out, diffTimestamp, diffTimestampFitsInBytes);
+ }
+
+ if ((flag & FLAG_SAME_TYPE) == 0) {
+ out.write(currentState.type);
+ }
+ ByteBufferUtils.skip(in, KeyValue.TIMESTAMP_TYPE_SIZE);
+
+ ByteBufferUtils.moveBufferToStream(out, in, valueLength);
+ }
+
+ private void uncompressSingleKeyValue(DataInputStream source,
+ ByteBuffer buffer,
+ DiffCompressionState state)
+ throws IOException, EncoderBufferTooSmallException {
+ // read the column family at the beginning
+ if (state.isFirst()) {
+ state.familyLength = source.readByte();
+ state.familyNameWithSize =
+ new byte[(state.familyLength & 0xff) + KeyValue.FAMILY_LENGTH_SIZE];
+ state.familyNameWithSize[0] = state.familyLength;
+ source.read(state.familyNameWithSize, KeyValue.FAMILY_LENGTH_SIZE,
+ state.familyLength);
+ }
+
+ // read flag
+ byte flag = source.readByte();
+
+ // read key/value/common lengths
+ int keyLength;
+ int valueLength;
+ if ((flag & FLAG_SAME_KEY_LENGTH) != 0) {
+ keyLength = state.keyLength;
+ } else {
+ keyLength = ByteBufferUtils.readCompressedInt(source);
+ }
+ if ((flag & FLAG_SAME_VALUE_LENGTH) != 0) {
+ valueLength = state.valueLength;
+ } else {
+ valueLength = ByteBufferUtils.readCompressedInt(source);
+ }
+ int commonPrefix = ByteBufferUtils.readCompressedInt(source);
+
+ // create KeyValue buffer and fill it prefix
+ int keyOffset = buffer.position();
+ ensureSpace(buffer, keyLength + valueLength + KeyValue.ROW_OFFSET);
+ buffer.putInt(keyLength);
+ buffer.putInt(valueLength);
+
+ // copy common from previous key
+ if (commonPrefix > 0) {
+ ByteBufferUtils.copyFromBufferToBuffer(buffer, buffer, state.prevOffset
+ + KeyValue.ROW_OFFSET, commonPrefix);
+ }
+
+ // copy the rest of the key from the buffer
+ int keyRestLength;
+ if (state.isFirst() || commonPrefix <
+ state.rowLength + KeyValue.ROW_LENGTH_SIZE) {
+ // omit the family part of the key, it is always the same
+ short rowLength;
+ int rowRestLength;
+
+ // check length of row
+ if (commonPrefix < KeyValue.ROW_LENGTH_SIZE) {
+ // not yet copied, do it now
+ ByteBufferUtils.copyFromStreamToBuffer(buffer, source,
+ KeyValue.ROW_LENGTH_SIZE - commonPrefix);
+ ByteBufferUtils.skip(buffer, -KeyValue.ROW_LENGTH_SIZE);
+ rowLength = buffer.getShort();
+ rowRestLength = rowLength;
+ } else {
+ // already in buffer, just read it
+ rowLength = buffer.getShort(keyOffset + KeyValue.ROW_OFFSET);
+ rowRestLength = rowLength + KeyValue.ROW_LENGTH_SIZE - commonPrefix;
+ }
+
+ // copy the rest of row
+ ByteBufferUtils.copyFromStreamToBuffer(buffer, source, rowRestLength);
+ state.rowLength = rowLength;
+
+ // copy the column family
+ buffer.put(state.familyNameWithSize);
+
+ keyRestLength = keyLength - rowLength -
+ state.familyNameWithSize.length -
+ (KeyValue.ROW_LENGTH_SIZE + KeyValue.TIMESTAMP_TYPE_SIZE);
+ } else {
+ // prevRowWithSizeLength is the same as on previous row
+ keyRestLength = keyLength - commonPrefix - KeyValue.TIMESTAMP_TYPE_SIZE;
+ }
+ // copy the rest of the key, after column family -> column qualifier
+ ByteBufferUtils.copyFromStreamToBuffer(buffer, source, keyRestLength);
+
+ // handle timestamp
+ int timestampFitsInBytes =
+ ((flag & MASK_TIMESTAMP_LENGTH) >>> SHIFT_TIMESTAMP_LENGTH) + 1;
+ long timestamp = ByteBufferUtils.readLong(source, timestampFitsInBytes);
+ if ((flag & FLAG_TIMESTAMP_SIGN) != 0) {
+ timestamp = -timestamp;
+ }
+ if ((flag & FLAG_TIMESTAMP_IS_DIFF) != 0) {
+ timestamp = state.timestamp - timestamp;
+ }
+ buffer.putLong(timestamp);
+
+ // copy the type field
+ byte type;
+ if ((flag & FLAG_SAME_TYPE) != 0) {
+ type = state.type;
+ } else {
+ type = source.readByte();
+ }
+ buffer.put(type);
+
+ // copy value part
+ ByteBufferUtils.copyFromStreamToBuffer(buffer, source, valueLength);
+
+ state.keyLength = keyLength;
+ state.valueLength = valueLength;
+ state.prevOffset = keyOffset;
+ state.timestamp = timestamp;
+ state.type = type;
+ // state.qualifier is unused
+ }
+
+ @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;
+ }
+ }
+
+ @Override
+ public ByteBuffer decodeKeyValues(DataInputStream source,
+ int allocHeaderLength, int skipLastBytes, boolean includesMemstoreTS)
+ throws IOException {
+ int decompressedSize = source.readInt();
+ ByteBuffer buffer = ByteBuffer.allocate(decompressedSize +
+ allocHeaderLength);
+ buffer.position(allocHeaderLength);
+ DiffCompressionState state = new DiffCompressionState();
+ while (source.available() > skipLastBytes) {
+ uncompressSingleKeyValue(source, buffer, state);
+ afterDecodingKeyValue(source, buffer, includesMemstoreTS);
+ }
+
+ if (source.available() != skipLastBytes) {
+ throw new IllegalStateException("Read too much bytes.");
+ }
+
+ return buffer;
+ }
+
+ @Override
+ public ByteBuffer getFirstKeyInBlock(ByteBuffer block) {
+ block.mark();
+ block.position(Bytes.SIZEOF_INT);
+ byte familyLength = block.get();
+ ByteBufferUtils.skip(block, familyLength);
+ byte flag = block.get();
+ int keyLength = ByteBufferUtils.readCompressedInt(block);
+ ByteBufferUtils.readCompressedInt(block); // valueLength
+ ByteBufferUtils.readCompressedInt(block); // commonLength
+ ByteBuffer result = ByteBuffer.allocate(keyLength);
+
+ // copy row
+ int pos = result.arrayOffset();
+ block.get(result.array(), pos, Bytes.SIZEOF_SHORT);
+ pos += Bytes.SIZEOF_SHORT;
+ short rowLength = result.getShort();
+ block.get(result.array(), pos, rowLength);
+ pos += rowLength;
+
+ // copy family
+ int savePosition = block.position();
+ block.position(Bytes.SIZEOF_INT);
+ block.get(result.array(), pos, familyLength + Bytes.SIZEOF_BYTE);
+ pos += familyLength + Bytes.SIZEOF_BYTE;
+
+ // copy qualifier
+ block.position(savePosition);
+ int qualifierLength =
+ keyLength - pos + result.arrayOffset() - KeyValue.TIMESTAMP_TYPE_SIZE;
+ block.get(result.array(), pos, qualifierLength);
+ pos += qualifierLength;
+
+ // copy the timestamp and type
+ int timestampFitInBytes =
+ ((flag & MASK_TIMESTAMP_LENGTH) >>> SHIFT_TIMESTAMP_LENGTH) + 1;
+ long timestamp = ByteBufferUtils.readLong(block, timestampFitInBytes);
+ if ((flag & FLAG_TIMESTAMP_SIGN) != 0) {
+ timestamp = -timestamp;
+ }
+ result.putLong(pos, timestamp);
+ pos += Bytes.SIZEOF_LONG;
+ block.get(result.array(), pos, Bytes.SIZEOF_BYTE);
+
+ block.reset();
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return DiffKeyDeltaEncoder.class.getSimpleName();
+ }
+
+ protected static class DiffSeekerState extends SeekerState {
+ private int rowLengthWithSize;
+ private long timestamp;
+
+ @Override
+ protected void copyFromNext(SeekerState that) {
+ super.copyFromNext(that);
+ DiffSeekerState other = (DiffSeekerState) that;
+ rowLengthWithSize = other.rowLengthWithSize;
+ timestamp = other.timestamp;
+ }
+ }
+
+ @Override
+ public EncodedSeeker createSeeker(RawComparator<byte[]> comparator,
+ final boolean includesMemstoreTS) {
+ return new BufferedEncodedSeeker<DiffSeekerState>(comparator) {
+ private byte[] familyNameWithSize;
+ private static final int TIMESTAMP_WITH_TYPE_LENGTH =
+ Bytes.SIZEOF_LONG + Bytes.SIZEOF_BYTE;
+
+ private void decode(boolean isFirst) {
+ byte flag = currentBuffer.get();
+ byte type = 0;
+ if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ if (!isFirst) {
+ type = current.keyBuffer[current.keyLength - Bytes.SIZEOF_BYTE];
+ }
+ current.keyLength = ByteBufferUtils.readCompressedInt(currentBuffer);
+ }
+ if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
+ current.valueLength =
+ ByteBufferUtils.readCompressedInt(currentBuffer);
+ }
+ current.lastCommonPrefix =
+ ByteBufferUtils.readCompressedInt(currentBuffer);
+
+ current.ensureSpaceForKey();
+
+ if (current.lastCommonPrefix < Bytes.SIZEOF_SHORT) {
+ // length of row is different, copy everything except family
+
+ // copy the row size
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ Bytes.SIZEOF_SHORT - current.lastCommonPrefix);
+ current.rowLengthWithSize = Bytes.toShort(current.keyBuffer, 0) +
+ Bytes.SIZEOF_SHORT;
+
+ // copy the rest of row
+ currentBuffer.get(current.keyBuffer, Bytes.SIZEOF_SHORT,
+ current.rowLengthWithSize - Bytes.SIZEOF_SHORT);
+
+ // copy the column family
+ System.arraycopy(familyNameWithSize, 0, current.keyBuffer,
+ current.rowLengthWithSize, familyNameWithSize.length);
+
+ // copy the qualifier
+ currentBuffer.get(current.keyBuffer,
+ current.rowLengthWithSize + familyNameWithSize.length,
+ current.keyLength - current.rowLengthWithSize -
+ familyNameWithSize.length - TIMESTAMP_WITH_TYPE_LENGTH);
+ } else if (current.lastCommonPrefix < current.rowLengthWithSize) {
+ // we have to copy part of row and qualifier,
+ // but column family is in right place
+
+ // before column family (rest of row)
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ current.rowLengthWithSize - current.lastCommonPrefix);
+
+ // after column family (qualifier)
+ currentBuffer.get(current.keyBuffer,
+ current.rowLengthWithSize + familyNameWithSize.length,
+ current.keyLength - current.rowLengthWithSize -
+ familyNameWithSize.length - TIMESTAMP_WITH_TYPE_LENGTH);
+ } else {
+ // copy just the ending
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ current.keyLength - TIMESTAMP_WITH_TYPE_LENGTH -
+ current.lastCommonPrefix);
+ }
+
+ // timestamp
+ int pos = current.keyLength - TIMESTAMP_WITH_TYPE_LENGTH;
+ int timestampFitInBytes = 1 +
+ ((flag & MASK_TIMESTAMP_LENGTH) >>> SHIFT_TIMESTAMP_LENGTH);
+ long timestampOrDiff =
+ ByteBufferUtils.readLong(currentBuffer, timestampFitInBytes);
+ if ((flag & FLAG_TIMESTAMP_SIGN) != 0) {
+ timestampOrDiff = -timestampOrDiff;
+ }
+ if ((flag & FLAG_TIMESTAMP_IS_DIFF) == 0) { // it is timestamp
+ current.timestamp = timestampOrDiff;
+ } else { // it is diff
+ current.timestamp = current.timestamp - timestampOrDiff;
+ }
+ Bytes.putLong(current.keyBuffer, pos, current.timestamp);
+ pos += Bytes.SIZEOF_LONG;
+
+ // type
+ if ((flag & FLAG_SAME_TYPE) == 0) {
+ currentBuffer.get(current.keyBuffer, pos, Bytes.SIZEOF_BYTE);
+ } else if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ current.keyBuffer[pos] = type;
+ }
+
+ current.valueOffset = currentBuffer.position();
+ ByteBufferUtils.skip(currentBuffer, current.valueLength);
+
+ if (includesMemstoreTS) {
+ current.memstoreTS = ByteBufferUtils.readVLong(currentBuffer);
+ } else {
+ current.memstoreTS = 0;
+ }
+ current.nextKvOffset = currentBuffer.position();
+ }
+
+ @Override
+ protected void decodeFirst() {
+ ByteBufferUtils.skip(currentBuffer, Bytes.SIZEOF_INT);
+
+ // read column family
+ byte familyNameLength = currentBuffer.get();
+ familyNameWithSize = new byte[familyNameLength + Bytes.SIZEOF_BYTE];
+ familyNameWithSize[0] = familyNameLength;
+ currentBuffer.get(familyNameWithSize, Bytes.SIZEOF_BYTE,
+ familyNameLength);
+ decode(true);
+ }
+
+ @Override
+ protected void decodeNext() {
+ decode(false);
+ }
+
+ @Override
+ protected DiffSeekerState createSeekerState() {
+ return new DiffSeekerState();
+ }
+ };
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodedDataBlock.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodedDataBlock.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodedDataBlock.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodedDataBlock.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,214 @@
+/*
+ * 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 java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.compress.Compressor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.io.NullOutputStream;
+
+/**
+ * Encapsulates a data block compressed using a particular encoding algorithm.
+ * Useful for testing and benchmarking.
+ */
+@InterfaceAudience.Private
+public class EncodedDataBlock {
+ private byte[] rawKVs;
+ private ByteBuffer rawBuffer;
+ private DataBlockEncoder dataBlockEncoder;
+
+ private byte[] cachedEncodedData;
+ private boolean includesMemstoreTS;
+
+ private final HFileBlockEncodingContext encodingCtx;
+
+ /**
+ * Create a buffer which will be encoded using dataBlockEncoder.
+ * @param dataBlockEncoder Algorithm used for compression.
+ * @param encoding encoding type used
+ * @param rawKVs
+ */
+ public EncodedDataBlock(DataBlockEncoder dataBlockEncoder,
+ boolean includesMemstoreTS, DataBlockEncoding encoding, byte[] rawKVs) {
+ Preconditions.checkNotNull(encoding,
+ "Cannot create encoded data block with null encoder");
+ this.dataBlockEncoder = dataBlockEncoder;
+ encodingCtx =
+ dataBlockEncoder.newDataBlockEncodingContext(Compression.Algorithm.NONE,
+ encoding, HConstants.HFILEBLOCK_DUMMY_HEADER);
+ this.rawKVs = rawKVs;
+ }
+
+ /**
+ * Provides access to compressed value.
+ * @return Forwards sequential iterator.
+ */
+ public Iterator<KeyValue> getIterator() {
+ final int rawSize = rawKVs.length;
+ byte[] encodedDataWithHeader = getEncodedData();
+ int bytesToSkip = encodingCtx.getHeaderSize() + Bytes.SIZEOF_SHORT;
+ ByteArrayInputStream bais = new ByteArrayInputStream(encodedDataWithHeader,
+ bytesToSkip, encodedDataWithHeader.length - bytesToSkip);
+ final DataInputStream dis = new DataInputStream(bais);
+
+ return new Iterator<KeyValue>() {
+ private ByteBuffer decompressedData = null;
+
+ @Override
+ public boolean hasNext() {
+ if (decompressedData == null) {
+ return rawSize > 0;
+ }
+ return decompressedData.hasRemaining();
+ }
+
+ @Override
+ public KeyValue next() {
+ if (decompressedData == null) {
+ try {
+ decompressedData = dataBlockEncoder.decodeKeyValues(
+ dis, includesMemstoreTS);
+ } catch (IOException e) {
+ throw new RuntimeException("Problem with data block encoder, " +
+ "most likely it requested more bytes than are available.", e);
+ }
+ decompressedData.rewind();
+ }
+
+ int offset = decompressedData.position();
+ KeyValue kv = new KeyValue(decompressedData.array(), offset);
+ decompressedData.position(offset + kv.getLength());
+
+ return kv;
+ }
+
+ @Override
+ public void remove() {
+ throw new NotImplementedException("remove() is not supported!");
+ }
+
+ @Override
+ public String toString() {
+ return "Iterator of: " + dataBlockEncoder.getClass().getName();
+ }
+
+ };
+ }
+
+ /**
+ * Find the size of minimal buffer that could store compressed data.
+ * @return Size in bytes of compressed data.
+ */
+ public int getSize() {
+ return getEncodedData().length;
+ }
+
+ /**
+ * Find the size of compressed data assuming that buffer will be compressed
+ * using given algorithm.
+ * @param algo compression algorithm
+ * @param compressor compressor already requested from codec
+ * @param inputBuffer Array to be compressed.
+ * @param offset Offset to beginning of the data.
+ * @param length Length to be compressed.
+ * @return Size of compressed data in bytes.
+ * @throws IOException
+ */
+ public static int getCompressedSize(Algorithm algo, Compressor compressor,
+ byte[] inputBuffer, int offset, int length) throws IOException {
+ DataOutputStream compressedStream = new DataOutputStream(
+ new NullOutputStream());
+ if (compressor != null) {
+ compressor.reset();
+ }
+ OutputStream compressingStream = algo.createCompressionStream(
+ compressedStream, compressor, 0);
+
+ compressingStream.write(inputBuffer, offset, length);
+ compressingStream.flush();
+ compressingStream.close();
+
+ return compressedStream.size();
+ }
+
+ /**
+ * Estimate size after second stage of compression (e.g. LZO).
+ * @param comprAlgo compression algorithm to be used for compression
+ * @param compressor compressor corresponding to the given compression
+ * algorithm
+ * @return Size after second stage of compression.
+ */
+ public int getEncodedCompressedSize(Algorithm comprAlgo,
+ Compressor compressor) throws IOException {
+ byte[] compressedBytes = getEncodedData();
+ return getCompressedSize(comprAlgo, compressor, compressedBytes, 0,
+ compressedBytes.length);
+ }
+
+ /** @return encoded data with header */
+ private byte[] getEncodedData() {
+ if (cachedEncodedData != null) {
+ return cachedEncodedData;
+ }
+ cachedEncodedData = encodeData();
+ return cachedEncodedData;
+ }
+
+ private ByteBuffer getUncompressedBuffer() {
+ if (rawBuffer == null || rawBuffer.limit() < rawKVs.length) {
+ rawBuffer = ByteBuffer.wrap(rawKVs);
+ }
+ return rawBuffer;
+ }
+
+ /**
+ * Do the encoding, but do not cache the encoded data.
+ * @return encoded data block with header and checksum
+ */
+ public byte[] encodeData() {
+ try {
+ this.dataBlockEncoder.encodeKeyValues(
+ getUncompressedBuffer(), includesMemstoreTS, encodingCtx);
+ } catch (IOException e) {
+ throw new RuntimeException(String.format(
+ "Bug in encoding part of algorithm %s. " +
+ "Probably it requested more bytes than are available.",
+ toString()), e);
+ }
+ return encodingCtx.getUncompressedBytesWithHeader();
+ }
+
+ @Override
+ public String toString() {
+ return dataBlockEncoder.toString();
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncoderBufferTooSmallException.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncoderBufferTooSmallException.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncoderBufferTooSmallException.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncoderBufferTooSmallException.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,31 @@
+/*
+ * 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.classification.InterfaceAudience;
+
+/**
+ * Internal error which indicates a bug in a data block encoding algorithm.
+ */
+@InterfaceAudience.Private
+public class EncoderBufferTooSmallException extends RuntimeException {
+ private static final long serialVersionUID = 4767495176134878737L;
+
+ public EncoderBufferTooSmallException(String message) {
+ super(message);
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,547 @@
+/*
+ * 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 java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.RawComparator;
+
+/**
+ * Encoder similar to {@link DiffKeyDeltaEncoder} but supposedly faster.
+ *
+ * Compress using:
+ * - store size of common prefix
+ * - save column family once in the first KeyValue
+ * - use integer compression for key, value and prefix (7-bit encoding)
+ * - use bits to avoid duplication key length, value length
+ * and type if it same as previous
+ * - store in 3 bits length of prefix timestamp
+ * with previous KeyValue's timestamp
+ * - one bit which allow to omit value if it is the same
+ *
+ * Format:
+ * - 1 byte: flag
+ * - 1-5 bytes: key length (only if FLAG_SAME_KEY_LENGTH is not set in flag)
+ * - 1-5 bytes: value length (only if FLAG_SAME_VALUE_LENGTH is not set in flag)
+ * - 1-5 bytes: prefix length
+ * - ... bytes: rest of the row (if prefix length is small enough)
+ * - ... bytes: qualifier (or suffix depending on prefix length)
+ * - 1-8 bytes: timestamp suffix
+ * - 1 byte: type (only if FLAG_SAME_TYPE is not set in the flag)
+ * - ... bytes: value (only if FLAG_SAME_VALUE is not set in the flag)
+ *
+ */
+@InterfaceAudience.Private
+public class FastDiffDeltaEncoder extends BufferedDataBlockEncoder {
+ final int MASK_TIMESTAMP_LENGTH = (1 << 0) | (1 << 1) | (1 << 2);
+ final int SHIFT_TIMESTAMP_LENGTH = 0;
+ final int FLAG_SAME_KEY_LENGTH = 1 << 3;
+ final int FLAG_SAME_VALUE_LENGTH = 1 << 4;
+ final int FLAG_SAME_TYPE = 1 << 5;
+ final int FLAG_SAME_VALUE = 1 << 6;
+
+ private static class FastDiffCompressionState extends CompressionState {
+ byte[] timestamp = new byte[KeyValue.TIMESTAMP_SIZE];
+ int prevTimestampOffset;
+
+ @Override
+ protected void readTimestamp(ByteBuffer in) {
+ in.get(timestamp);
+ }
+
+ @Override
+ void copyFrom(CompressionState state) {
+ super.copyFrom(state);
+ FastDiffCompressionState state2 = (FastDiffCompressionState) state;
+ System.arraycopy(state2.timestamp, 0, timestamp, 0,
+ KeyValue.TIMESTAMP_SIZE);
+ prevTimestampOffset = state2.prevTimestampOffset;
+ }
+
+ /**
+ * Copies the first key/value from the given stream, and initializes
+ * decompression state based on it. Assumes that we have already read key
+ * and value lengths. Does not set {@link #qualifierLength} (not used by
+ * decompression) or {@link #prevOffset} (set by the calle afterwards).
+ */
+ private void decompressFirstKV(ByteBuffer out, DataInputStream in)
+ throws IOException {
+ int kvPos = out.position();
+ out.putInt(keyLength);
+ out.putInt(valueLength);
+ prevTimestampOffset = out.position() + keyLength -
+ KeyValue.TIMESTAMP_TYPE_SIZE;
+ ByteBufferUtils.copyFromStreamToBuffer(out, in, keyLength + valueLength);
+ rowLength = out.getShort(kvPos + KeyValue.ROW_OFFSET);
+ familyLength = out.get(kvPos + KeyValue.ROW_OFFSET +
+ KeyValue.ROW_LENGTH_SIZE + rowLength);
+ type = out.get(prevTimestampOffset + KeyValue.TIMESTAMP_SIZE);
+ }
+
+ }
+
+ private void compressSingleKeyValue(
+ FastDiffCompressionState previousState,
+ FastDiffCompressionState currentState,
+ OutputStream out, ByteBuffer in) throws IOException {
+ currentState.prevOffset = in.position();
+ int keyLength = in.getInt();
+ int valueOffset =
+ currentState.prevOffset + keyLength + KeyValue.ROW_OFFSET;
+ int valueLength = in.getInt();
+ byte flag = 0;
+
+ if (previousState.isFirst()) {
+ // copy the key, there is no common prefix with none
+ out.write(flag);
+ ByteBufferUtils.putCompressedInt(out, keyLength);
+ ByteBufferUtils.putCompressedInt(out, valueLength);
+ ByteBufferUtils.putCompressedInt(out, 0);
+
+ currentState.readKey(in, keyLength, valueLength);
+
+ ByteBufferUtils.moveBufferToStream(out, in, keyLength + valueLength);
+ } else {
+ // find a common prefix and skip it
+ int commonPrefix = ByteBufferUtils.findCommonPrefix(in, in.position(),
+ previousState.prevOffset + KeyValue.ROW_OFFSET,
+ Math.min(keyLength, previousState.keyLength) -
+ KeyValue.TIMESTAMP_TYPE_SIZE);
+
+ currentState.readKey(in, keyLength, valueLength,
+ commonPrefix, previousState);
+
+ if (keyLength == previousState.keyLength) {
+ flag |= FLAG_SAME_KEY_LENGTH;
+ }
+ if (valueLength == previousState.valueLength) {
+ flag |= FLAG_SAME_VALUE_LENGTH;
+ }
+ if (currentState.type == previousState.type) {
+ flag |= FLAG_SAME_TYPE;
+ }
+
+ int commonTimestampPrefix = findCommonTimestampPrefix(
+ currentState, previousState);
+ flag |= commonTimestampPrefix << SHIFT_TIMESTAMP_LENGTH;
+
+ // Check if current and previous values are the same. Compare value
+ // length first as an optimization.
+ if (valueLength == previousState.valueLength) {
+ int previousValueOffset = previousState.prevOffset
+ + previousState.keyLength + KeyValue.ROW_OFFSET;
+ if (ByteBufferUtils.arePartsEqual(in,
+ previousValueOffset, previousState.valueLength,
+ valueOffset, valueLength)) {
+ flag |= FLAG_SAME_VALUE;
+ }
+ }
+
+ out.write(flag);
+ if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ ByteBufferUtils.putCompressedInt(out, keyLength);
+ }
+ if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
+ ByteBufferUtils.putCompressedInt(out, valueLength);
+ }
+ ByteBufferUtils.putCompressedInt(out, commonPrefix);
+
+ ByteBufferUtils.skip(in, commonPrefix);
+ if (commonPrefix < currentState.rowLength + KeyValue.ROW_LENGTH_SIZE) {
+ // Previous and current rows are different. Copy the differing part of
+ // the row, skip the column family, and copy the qualifier.
+ ByteBufferUtils.moveBufferToStream(out, in,
+ currentState.rowLength + KeyValue.ROW_LENGTH_SIZE - commonPrefix);
+ ByteBufferUtils.skip(in, currentState.familyLength +
+ KeyValue.FAMILY_LENGTH_SIZE);
+ ByteBufferUtils.moveBufferToStream(out, in,
+ currentState.qualifierLength);
+ } else {
+ // The common part includes the whole row. As the column family is the
+ // same across the whole file, it will automatically be included in the
+ // common prefix, so we need not special-case it here.
+ int restKeyLength = keyLength - commonPrefix -
+ KeyValue.TIMESTAMP_TYPE_SIZE;
+ ByteBufferUtils.moveBufferToStream(out, in, restKeyLength);
+ }
+ ByteBufferUtils.skip(in, commonTimestampPrefix);
+ ByteBufferUtils.moveBufferToStream(out, in,
+ KeyValue.TIMESTAMP_SIZE - commonTimestampPrefix);
+
+ // Write the type if it is not the same as before.
+ if ((flag & FLAG_SAME_TYPE) == 0) {
+ out.write(currentState.type);
+ }
+
+ // Write the value if it is not the same as before.
+ if ((flag & FLAG_SAME_VALUE) == 0) {
+ ByteBufferUtils.copyBufferToStream(out, in, valueOffset, valueLength);
+ }
+
+ // Skip key type and value in the input buffer.
+ ByteBufferUtils.skip(in, KeyValue.TYPE_SIZE + currentState.valueLength);
+ }
+ }
+
+ private int findCommonTimestampPrefix(FastDiffCompressionState left,
+ FastDiffCompressionState right) {
+ int prefixTimestamp = 0;
+ while (prefixTimestamp < (KeyValue.TIMESTAMP_SIZE - 1) &&
+ left.timestamp[prefixTimestamp]
+ == right.timestamp[prefixTimestamp]) {
+ prefixTimestamp++;
+ }
+ return prefixTimestamp; // has to be at most 7 bytes
+ }
+
+ private void uncompressSingleKeyValue(DataInputStream source,
+ ByteBuffer out, FastDiffCompressionState state)
+ throws IOException, EncoderBufferTooSmallException {
+ byte flag = source.readByte();
+ int prevKeyLength = state.keyLength;
+
+ if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ state.keyLength = ByteBufferUtils.readCompressedInt(source);
+ }
+ if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
+ state.valueLength = ByteBufferUtils.readCompressedInt(source);
+ }
+ int commonLength = ByteBufferUtils.readCompressedInt(source);
+
+ ensureSpace(out, state.keyLength + state.valueLength + KeyValue.ROW_OFFSET);
+
+ int kvPos = out.position();
+
+ if (!state.isFirst()) {
+ // copy the prefix
+ int common;
+ int prevOffset;
+
+ if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
+ out.putInt(state.keyLength);
+ out.putInt(state.valueLength);
+ prevOffset = state.prevOffset + KeyValue.ROW_OFFSET;
+ common = commonLength;
+ } else {
+ if ((flag & FLAG_SAME_KEY_LENGTH) != 0) {
+ prevOffset = state.prevOffset;
+ common = commonLength + KeyValue.ROW_OFFSET;
+ } else {
+ out.putInt(state.keyLength);
+ prevOffset = state.prevOffset + KeyValue.KEY_LENGTH_SIZE;
+ common = commonLength + KeyValue.KEY_LENGTH_SIZE;
+ }
+ }
+
+ ByteBufferUtils.copyFromBufferToBuffer(out, out, prevOffset, common);
+
+ // copy the rest of the key from the buffer
+ int keyRestLength;
+ if (commonLength < state.rowLength + KeyValue.ROW_LENGTH_SIZE) {
+ // omit the family part of the key, it is always the same
+ int rowWithSizeLength;
+ int rowRestLength;
+
+ // check length of row
+ if (commonLength < KeyValue.ROW_LENGTH_SIZE) {
+ // not yet copied, do it now
+ ByteBufferUtils.copyFromStreamToBuffer(out, source,
+ KeyValue.ROW_LENGTH_SIZE - commonLength);
+
+ rowWithSizeLength = out.getShort(out.position() -
+ KeyValue.ROW_LENGTH_SIZE) + KeyValue.ROW_LENGTH_SIZE;
+ rowRestLength = rowWithSizeLength - KeyValue.ROW_LENGTH_SIZE;
+ } else {
+ // already in kvBuffer, just read it
+ rowWithSizeLength = out.getShort(kvPos + KeyValue.ROW_OFFSET) +
+ KeyValue.ROW_LENGTH_SIZE;
+ rowRestLength = rowWithSizeLength - commonLength;
+ }
+
+ // copy the rest of row
+ ByteBufferUtils.copyFromStreamToBuffer(out, source, rowRestLength);
+
+ // copy the column family
+ ByteBufferUtils.copyFromBufferToBuffer(out, out,
+ state.prevOffset + KeyValue.ROW_OFFSET + KeyValue.ROW_LENGTH_SIZE
+ + state.rowLength, state.familyLength
+ + KeyValue.FAMILY_LENGTH_SIZE);
+ state.rowLength = (short) (rowWithSizeLength -
+ KeyValue.ROW_LENGTH_SIZE);
+
+ keyRestLength = state.keyLength - rowWithSizeLength -
+ state.familyLength -
+ (KeyValue.FAMILY_LENGTH_SIZE + KeyValue.TIMESTAMP_TYPE_SIZE);
+ } else {
+ // prevRowWithSizeLength is the same as on previous row
+ keyRestLength = state.keyLength - commonLength -
+ KeyValue.TIMESTAMP_TYPE_SIZE;
+ }
+ // copy the rest of the key, after column family == column qualifier
+ ByteBufferUtils.copyFromStreamToBuffer(out, source, keyRestLength);
+
+ // copy timestamp
+ int prefixTimestamp =
+ (flag & MASK_TIMESTAMP_LENGTH) >>> SHIFT_TIMESTAMP_LENGTH;
+ ByteBufferUtils.copyFromBufferToBuffer(out, out,
+ state.prevTimestampOffset, prefixTimestamp);
+ state.prevTimestampOffset = out.position() - prefixTimestamp;
+ ByteBufferUtils.copyFromStreamToBuffer(out, source,
+ KeyValue.TIMESTAMP_SIZE - prefixTimestamp);
+
+ // copy the type and value
+ if ((flag & FLAG_SAME_TYPE) != 0) {
+ out.put(state.type);
+ if ((flag & FLAG_SAME_VALUE) != 0) {
+ ByteBufferUtils.copyFromBufferToBuffer(out, out, state.prevOffset +
+ KeyValue.ROW_OFFSET + prevKeyLength, state.valueLength);
+ } else {
+ ByteBufferUtils.copyFromStreamToBuffer(out, source,
+ state.valueLength);
+ }
+ } else {
+ if ((flag & FLAG_SAME_VALUE) != 0) {
+ ByteBufferUtils.copyFromStreamToBuffer(out, source,
+ KeyValue.TYPE_SIZE);
+ ByteBufferUtils.copyFromBufferToBuffer(out, out, state.prevOffset +
+ KeyValue.ROW_OFFSET + prevKeyLength, state.valueLength);
+ } else {
+ ByteBufferUtils.copyFromStreamToBuffer(out, source,
+ state.valueLength + KeyValue.TYPE_SIZE);
+ }
+ state.type = out.get(state.prevTimestampOffset +
+ KeyValue.TIMESTAMP_SIZE);
+ }
+ } else { // this is the first element
+ state.decompressFirstKV(out, source);
+ }
+
+ state.prevOffset = kvPos;
+ }
+
+ @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;
+ }
+ }
+
+ @Override
+ public ByteBuffer decodeKeyValues(DataInputStream source,
+ int allocHeaderLength, int skipLastBytes, boolean includesMemstoreTS)
+ throws IOException {
+ int decompressedSize = source.readInt();
+ ByteBuffer buffer = ByteBuffer.allocate(decompressedSize +
+ allocHeaderLength);
+ buffer.position(allocHeaderLength);
+ FastDiffCompressionState state = new FastDiffCompressionState();
+ while (source.available() > skipLastBytes) {
+ uncompressSingleKeyValue(source, buffer, state);
+ afterDecodingKeyValue(source, buffer, includesMemstoreTS);
+ }
+
+ if (source.available() != skipLastBytes) {
+ throw new IllegalStateException("Read too much bytes.");
+ }
+
+ return buffer;
+ }
+
+ @Override
+ public ByteBuffer getFirstKeyInBlock(ByteBuffer block) {
+ block.mark();
+ block.position(Bytes.SIZEOF_INT + Bytes.SIZEOF_BYTE);
+ int keyLength = ByteBufferUtils.readCompressedInt(block);
+ ByteBufferUtils.readCompressedInt(block); // valueLength
+ ByteBufferUtils.readCompressedInt(block); // commonLength
+ int pos = block.position();
+ block.reset();
+ return ByteBuffer.wrap(block.array(), pos, keyLength).slice();
+ }
+
+ @Override
+ public String toString() {
+ return FastDiffDeltaEncoder.class.getSimpleName();
+ }
+
+ protected static class FastDiffSeekerState extends SeekerState {
+ private byte[] prevTimestampAndType =
+ new byte[KeyValue.TIMESTAMP_TYPE_SIZE];
+ private int rowLengthWithSize;
+ private int familyLengthWithSize;
+
+ @Override
+ protected void copyFromNext(SeekerState that) {
+ super.copyFromNext(that);
+ FastDiffSeekerState other = (FastDiffSeekerState) that;
+ System.arraycopy(other.prevTimestampAndType, 0,
+ prevTimestampAndType, 0,
+ KeyValue.TIMESTAMP_TYPE_SIZE);
+ rowLengthWithSize = other.rowLengthWithSize;
+ familyLengthWithSize = other.familyLengthWithSize;
+ }
+ }
+
+ @Override
+ public EncodedSeeker createSeeker(RawComparator<byte[]> comparator,
+ final boolean includesMemstoreTS) {
+ return new BufferedEncodedSeeker<FastDiffSeekerState>(comparator) {
+ private void decode(boolean isFirst) {
+ byte flag = currentBuffer.get();
+ if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ if (!isFirst) {
+ System.arraycopy(current.keyBuffer,
+ current.keyLength - current.prevTimestampAndType.length,
+ current.prevTimestampAndType, 0,
+ current.prevTimestampAndType.length);
+ }
+ current.keyLength = ByteBufferUtils.readCompressedInt(currentBuffer);
+ }
+ if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
+ current.valueLength =
+ ByteBufferUtils.readCompressedInt(currentBuffer);
+ }
+ current.lastCommonPrefix =
+ ByteBufferUtils.readCompressedInt(currentBuffer);
+
+ current.ensureSpaceForKey();
+
+ if (isFirst) {
+ // copy everything
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ current.keyLength - current.prevTimestampAndType.length);
+ current.rowLengthWithSize = Bytes.toShort(current.keyBuffer, 0) +
+ Bytes.SIZEOF_SHORT;
+ current.familyLengthWithSize =
+ current.keyBuffer[current.rowLengthWithSize] + Bytes.SIZEOF_BYTE;
+ } else if (current.lastCommonPrefix < Bytes.SIZEOF_SHORT) {
+ // length of row is different, copy everything except family
+
+ // copy the row size
+ int oldRowLengthWithSize = current.rowLengthWithSize;
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ Bytes.SIZEOF_SHORT - current.lastCommonPrefix);
+ current.rowLengthWithSize = Bytes.toShort(current.keyBuffer, 0) +
+ Bytes.SIZEOF_SHORT;
+
+ // move the column family
+ System.arraycopy(current.keyBuffer, oldRowLengthWithSize,
+ current.keyBuffer, current.rowLengthWithSize,
+ current.familyLengthWithSize);
+
+ // copy the rest of row
+ currentBuffer.get(current.keyBuffer, Bytes.SIZEOF_SHORT,
+ current.rowLengthWithSize - Bytes.SIZEOF_SHORT);
+
+ // copy the qualifier
+ currentBuffer.get(current.keyBuffer, current.rowLengthWithSize
+ + current.familyLengthWithSize, current.keyLength
+ - current.rowLengthWithSize - current.familyLengthWithSize
+ - current.prevTimestampAndType.length);
+ } else if (current.lastCommonPrefix < current.rowLengthWithSize) {
+ // We have to copy part of row and qualifier, but the column family
+ // is in the right place.
+
+ // before column family (rest of row)
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ current.rowLengthWithSize - current.lastCommonPrefix);
+
+ // after column family (qualifier)
+ currentBuffer.get(current.keyBuffer, current.rowLengthWithSize
+ + current.familyLengthWithSize, current.keyLength
+ - current.rowLengthWithSize - current.familyLengthWithSize
+ - current.prevTimestampAndType.length);
+ } else {
+ // copy just the ending
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ current.keyLength - current.prevTimestampAndType.length
+ - current.lastCommonPrefix);
+ }
+
+ // timestamp
+ int pos = current.keyLength - current.prevTimestampAndType.length;
+ int commonTimestampPrefix = (flag & MASK_TIMESTAMP_LENGTH) >>>
+ SHIFT_TIMESTAMP_LENGTH;
+ if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ System.arraycopy(current.prevTimestampAndType, 0, current.keyBuffer,
+ pos, commonTimestampPrefix);
+ }
+ pos += commonTimestampPrefix;
+ currentBuffer.get(current.keyBuffer, pos,
+ Bytes.SIZEOF_LONG - commonTimestampPrefix);
+ pos += Bytes.SIZEOF_LONG - commonTimestampPrefix;
+
+ // type
+ if ((flag & FLAG_SAME_TYPE) == 0) {
+ currentBuffer.get(current.keyBuffer, pos, Bytes.SIZEOF_BYTE);
+ } else if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
+ current.keyBuffer[pos] =
+ current.prevTimestampAndType[Bytes.SIZEOF_LONG];
+ }
+
+ // handle value
+ if ((flag & FLAG_SAME_VALUE) == 0) {
+ current.valueOffset = currentBuffer.position();
+ ByteBufferUtils.skip(currentBuffer, current.valueLength);
+ }
+
+ if (includesMemstoreTS) {
+ current.memstoreTS = ByteBufferUtils.readVLong(currentBuffer);
+ } else {
+ current.memstoreTS = 0;
+ }
+ current.nextKvOffset = currentBuffer.position();
+ }
+
+ @Override
+ protected void decodeFirst() {
+ ByteBufferUtils.skip(currentBuffer, Bytes.SIZEOF_INT);
+ decode(true);
+ }
+
+ @Override
+ protected void decodeNext() {
+ decode(false);
+ }
+
+ @Override
+ protected FastDiffSeekerState createSeekerState() {
+ return new FastDiffSeekerState();
+ }
+ };
+ }
+}
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultEncodingContext.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultEncodingContext.java?rev=1449950&r1=1449949&r2=1449950&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultEncodingContext.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultEncodingContext.java Mon Feb 25 22:50:17 2013
@@ -87,7 +87,7 @@ public class HFileBlockDefaultEncodingCo
}
}
dummyHeader = Preconditions.checkNotNull(headerBytes,
- "Please pass HFileBlock.DUMMY_HEADER instead of null for param headerBytes");
+ "Please pass HFileBlock.HFILEBLOCK_DUMMY_HEADER instead of null for param headerBytes");
}
/**
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,201 @@
+/*
+ * 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 java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.RawComparator;
+
+/**
+ * Compress key by storing size of common prefix with previous KeyValue
+ * and storing raw size of rest.
+ *
+ * Format:
+ * 1-5 bytes: compressed key length minus prefix (7-bit encoding)
+ * 1-5 bytes: compressed value length (7-bit encoding)
+ * 1-3 bytes: compressed length of common key prefix
+ * ... bytes: rest of key (including timestamp)
+ * ... bytes: value
+ *
+ * In a worst case compressed KeyValue will be three bytes longer than original.
+ *
+ */
+@InterfaceAudience.Private
+public class PrefixKeyDeltaEncoder extends BufferedDataBlockEncoder {
+
+ private int addKV(int prevKeyOffset, DataOutputStream out,
+ ByteBuffer in, int prevKeyLength) throws IOException {
+ int keyLength = in.getInt();
+ int valueLength = in.getInt();
+
+ if (prevKeyOffset == -1) {
+ // copy the key, there is no common prefix with none
+ ByteBufferUtils.putCompressedInt(out, keyLength);
+ ByteBufferUtils.putCompressedInt(out, valueLength);
+ ByteBufferUtils.putCompressedInt(out, 0);
+ ByteBufferUtils.moveBufferToStream(out, in, keyLength + valueLength);
+ } else {
+ // find a common prefix and skip it
+ int common = ByteBufferUtils.findCommonPrefix(
+ in, prevKeyOffset + KeyValue.ROW_OFFSET,
+ in.position(),
+ Math.min(prevKeyLength, keyLength));
+
+ ByteBufferUtils.putCompressedInt(out, keyLength - common);
+ ByteBufferUtils.putCompressedInt(out, valueLength);
+ ByteBufferUtils.putCompressedInt(out, common);
+
+ ByteBufferUtils.skip(in, common);
+ ByteBufferUtils.moveBufferToStream(out, in, keyLength - common
+ + valueLength);
+ }
+
+ return keyLength;
+ }
+
+ @Override
+ public void internalEncodeKeyValues(DataOutputStream writeHere,
+ ByteBuffer in, boolean includesMemstoreTS) throws IOException {
+ in.rewind();
+ ByteBufferUtils.putInt(writeHere, in.limit());
+ int prevOffset = -1;
+ int offset = 0;
+ int keyLength = 0;
+ while (in.hasRemaining()) {
+ offset = in.position();
+ keyLength = addKV(prevOffset, writeHere, in, keyLength);
+ afterEncodingKeyValue(in, writeHere, includesMemstoreTS);
+ prevOffset = offset;
+ }
+ }
+
+ @Override
+ public ByteBuffer decodeKeyValues(DataInputStream source,
+ int allocHeaderLength, int skipLastBytes, boolean includesMemstoreTS)
+ throws IOException {
+ int decompressedSize = source.readInt();
+ ByteBuffer buffer = ByteBuffer.allocate(decompressedSize +
+ allocHeaderLength);
+ buffer.position(allocHeaderLength);
+ int prevKeyOffset = 0;
+
+ while (source.available() > skipLastBytes) {
+ prevKeyOffset = decodeKeyValue(source, buffer, prevKeyOffset);
+ afterDecodingKeyValue(source, buffer, includesMemstoreTS);
+ }
+
+ if (source.available() != skipLastBytes) {
+ throw new IllegalStateException("Read too many bytes.");
+ }
+
+ buffer.limit(buffer.position());
+ return buffer;
+ }
+
+ private int decodeKeyValue(DataInputStream source, ByteBuffer buffer,
+ int prevKeyOffset)
+ throws IOException, EncoderBufferTooSmallException {
+ int keyLength = ByteBufferUtils.readCompressedInt(source);
+ int valueLength = ByteBufferUtils.readCompressedInt(source);
+ int commonLength = ByteBufferUtils.readCompressedInt(source);
+ int keyOffset;
+ keyLength += commonLength;
+
+ ensureSpace(buffer, keyLength + valueLength + KeyValue.ROW_OFFSET);
+
+ buffer.putInt(keyLength);
+ buffer.putInt(valueLength);
+
+ // copy the prefix
+ if (commonLength > 0) {
+ keyOffset = buffer.position();
+ ByteBufferUtils.copyFromBufferToBuffer(buffer, buffer, prevKeyOffset,
+ commonLength);
+ } else {
+ keyOffset = buffer.position();
+ }
+
+ // copy rest of the key and value
+ int len = keyLength - commonLength + valueLength;
+ ByteBufferUtils.copyFromStreamToBuffer(buffer, source, len);
+ return keyOffset;
+ }
+
+ @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(), pos, keyLength).slice();
+ }
+
+ @Override
+ public String toString() {
+ return PrefixKeyDeltaEncoder.class.getSimpleName();
+ }
+
+ @Override
+ public EncodedSeeker createSeeker(RawComparator<byte[]> comparator,
+ final boolean includesMemstoreTS) {
+ return new BufferedEncodedSeeker<SeekerState>(comparator) {
+ @Override
+ protected void decodeNext() {
+ current.keyLength = ByteBufferUtils.readCompressedInt(currentBuffer);
+ current.valueLength = ByteBufferUtils.readCompressedInt(currentBuffer);
+ current.lastCommonPrefix =
+ ByteBufferUtils.readCompressedInt(currentBuffer);
+ current.keyLength += current.lastCommonPrefix;
+ current.ensureSpaceForKey();
+ currentBuffer.get(current.keyBuffer, current.lastCommonPrefix,
+ current.keyLength - current.lastCommonPrefix);
+ current.valueOffset = currentBuffer.position();
+ ByteBufferUtils.skip(currentBuffer, current.valueLength);
+ if (includesMemstoreTS) {
+ current.memstoreTS = ByteBufferUtils.readVLong(currentBuffer);
+ } else {
+ current.memstoreTS = 0;
+ }
+ current.nextKvOffset = currentBuffer.position();
+ }
+
+ @Override
+ protected void decodeFirst() {
+ ByteBufferUtils.skip(currentBuffer, Bytes.SIZEOF_INT);
+ decodeNext();
+ }
+ };
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Addressing.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Addressing.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Addressing.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Addressing.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,80 @@
+/**
+ *
+ * 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.util;
+
+import java.net.InetSocketAddress;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Utility for network addresses, resolving and naming.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class Addressing {
+ public static final String VALID_PORT_REGEX = "[\\d]+";
+ public static final String HOSTNAME_PORT_SEPARATOR = ":";
+
+ /**
+ * @param hostAndPort Formatted as <code><hostname> ':' <port></code>
+ * @return An InetSocketInstance
+ */
+ public static InetSocketAddress createInetSocketAddressFromHostAndPortStr(
+ final String hostAndPort) {
+ return new InetSocketAddress(parseHostname(hostAndPort), parsePort(hostAndPort));
+ }
+
+ /**
+ * @param hostname Server hostname
+ * @param port Server port
+ * @return Returns a concatenation of <code>hostname</code> and
+ * <code>port</code> in following
+ * form: <code><hostname> ':' <port></code>. For example, if hostname
+ * is <code>example.org</code> and port is 1234, this method will return
+ * <code>example.org:1234</code>
+ */
+ public static String createHostAndPortStr(final String hostname, final int port) {
+ return hostname + HOSTNAME_PORT_SEPARATOR + port;
+ }
+
+ /**
+ * @param hostAndPort Formatted as <code><hostname> ':' <port></code>
+ * @return The hostname portion of <code>hostAndPort</code>
+ */
+ public static String parseHostname(final String hostAndPort) {
+ int colonIndex = hostAndPort.lastIndexOf(HOSTNAME_PORT_SEPARATOR);
+ if (colonIndex < 0) {
+ throw new IllegalArgumentException("Not a host:port pair: " + hostAndPort);
+ }
+ return hostAndPort.substring(0, colonIndex);
+ }
+
+ /**
+ * @param hostAndPort Formatted as <code><hostname> ':' <port></code>
+ * @return The port portion of <code>hostAndPort</code>
+ */
+ public static int parsePort(final String hostAndPort) {
+ int colonIndex = hostAndPort.lastIndexOf(HOSTNAME_PORT_SEPARATOR);
+ if (colonIndex < 0) {
+ throw new IllegalArgumentException("Not a host:port pair: " + hostAndPort);
+ }
+ return Integer.parseInt(hostAndPort.substring(colonIndex + 1));
+ }
+}