You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2013/02/23 23:38:59 UTC
svn commit: r1449420 [1/2] - in /hbase/trunk: hbase-common/
hbase-common/src/main/java/org/apache/hadoop/hbase/
hbase-common/src/main/java/org/apache/hadoop/hbase/io/
hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/
hbase-common/src/main...
Author: stack
Date: Sat Feb 23 22:38:58 2013
New Revision: 1449420
URL: http://svn.apache.org/r1449420
Log:
HBASE-7899 Cell block building tools: Cell codec and means of iterating an objects Cells
Added:
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueUtil.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellComparator.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScannable.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScanner.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellUtil.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseDecoder.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseEncoder.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CellCodec.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/Codec.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/KeyValueCodec.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/io/
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java
hbase/trunk/hbase-common/src/test/java/org/apache/hbase/
hbase/trunk/hbase-common/src/test/java/org/apache/hbase/TestCellUtil.java
hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/
hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java
hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestKeyValueCodec.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/scanner/CellScannerPosition.java
Removed:
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/cell/
Modified:
hbase/trunk/hbase-common/pom.xml
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTestUtil.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ImmutableBytesWritable.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/JVM.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/LoadTestKVGenerator.java
hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/RedundantKVGenerator.java
hbase/trunk/hbase-common/src/main/java/org/apache/hbase/Cell.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/PrefixTreeCodec.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/PrefixTreeSeeker.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/decode/PrefixTreeArrayScanner.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/decode/PrefixTreeArraySearcher.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/decode/PrefixTreeCell.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/encode/PrefixTreeEncoder.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/scanner/CellScanner.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/scanner/CellSearcher.java
hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/scanner/ReversibleCellScanner.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/BaseTestRowData.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/TestPrefixTreeSearcher.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/TestRowEncoder.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/data/TestRowDataDeeper.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/data/TestRowDataNumberStrings.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/data/TestRowDataSearcherRowMiss.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/data/TestRowDataSimple.java
hbase/trunk/hbase-prefix-tree/src/test/java/org/apache/hbase/codec/prefixtree/row/data/TestRowDataTrivial.java
hbase/trunk/hbase-protocol/pom.xml
hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ZooKeeperProtos.java
hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestColumnSeeking.java
hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMultiColumnScanner.java
Modified: hbase/trunk/hbase-common/pom.xml
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/pom.xml?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/pom.xml (original)
+++ hbase/trunk/hbase-common/pom.xml Sat Feb 23 22:38:58 2013
@@ -77,7 +77,7 @@
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
- </dependency>
+ </dependency>
</dependencies>
<profiles>
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java Sat Feb 23 22:38:58 2013
@@ -22,6 +22,8 @@ package org.apache.hadoop.hbase;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.util.Comparator;
@@ -35,9 +37,10 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
+import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.RawComparator;
import org.apache.hbase.Cell;
-import org.apache.hbase.cell.CellComparator;
+import org.apache.hbase.CellComparator;
import com.google.common.primitives.Longs;
@@ -64,7 +67,7 @@ import com.google.common.primitives.Long
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
-public class KeyValue implements Cell, HeapSize {
+public class KeyValue implements Cell, HeapSize, Cloneable {
static final Log LOG = LogFactory.getLog(KeyValue.class);
// TODO: Group Key-only comparators and operations into a Key class, just
// for neatness sake, if can figure what to call it.
@@ -915,7 +918,7 @@ public class KeyValue implements Cell, H
return "empty";
}
return keyToString(this.bytes, this.offset + ROW_OFFSET, getKeyLength()) +
- "/vlen=" + getValueLength() + "/ts=" + memstoreTS;
+ "/vlen=" + getValueLength() + "/mvcc=" + memstoreTS;
}
/**
@@ -2299,8 +2302,10 @@ public class KeyValue implements Cell, H
}
/**
- * @param in Where to read bytes from
- * @return KeyValue created by deserializing from <code>in</code>
+ * @param in Where to read bytes from. Creates a byte array to hold the KeyValue
+ * backing bytes copied from the steam.
+ * @return KeyValue created by deserializing from <code>in</code> OR if we find a length
+ * of zero, we will return null which can be useful marking a stream as done.
* @throws IOException
*/
public static KeyValue create(final DataInput in) throws IOException {
@@ -2311,10 +2316,12 @@ public class KeyValue implements Cell, H
* Create a KeyValue reading <code>length</code> from <code>in</code>
* @param length
* @param in
- * @return Created KeyValue
+ * @return Created KeyValue OR if we find a length of zero, we will return null which
+ * can be useful marking a stream as done.
* @throws IOException
*/
public static KeyValue create(int length, final DataInput in) throws IOException {
+ if (length == 0) return null;
// This is how the old Writables.readFrom used to deserialize. Didn't even vint.
byte [] bytes = new byte[length];
in.readFully(bytes);
@@ -2322,6 +2329,24 @@ public class KeyValue implements Cell, H
}
/**
+ * Create a KeyValue reading from the raw InputStream.
+ * Named <code>iscreate</code> so doesn't clash with {@link #create(DataInput)}
+ * @param in
+ * @return Created KeyValue OR if we find a length of zero, we will return null which
+ * can be useful marking a stream as done.
+ * @throws IOException
+ */
+ public static KeyValue iscreate(final InputStream in) throws IOException {
+ byte [] intBytes = new byte[Bytes.SIZEOF_INT];
+ int length = in.read(intBytes);
+ if (length == 0) return null;
+ if (length != intBytes.length) throw new IOException("Failed read of int length " + length);
+ byte [] bytes = new byte[Bytes.toInt(intBytes)];
+ IOUtils.readFully(in, bytes, 0, bytes.length);
+ return new KeyValue(bytes, 0, bytes.length);
+ }
+
+ /**
* Write out a KeyValue in the manner in which we used to when KeyValue was a Writable.
* @param kv
* @param out
@@ -2330,8 +2355,8 @@ public class KeyValue implements Cell, H
* @see #create(DataInput) for the inverse function
*/
public static long write(final KeyValue kv, final DataOutput out) throws IOException {
- // This is how the old Writables write used to serialize KVs. Need to figure way to make it work for all
- // implementations.
+ // This is how the old Writables write used to serialize KVs. Need to figure way to make it
+ // work for all implementations.
int length = kv.getLength();
out.writeInt(length);
out.write(kv.getBuffer(), kv.getOffset(), length);
@@ -2339,6 +2364,25 @@ public class KeyValue implements Cell, H
}
/**
+ * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable but do
+ * not require a {@link DataOutput}, just take plain {@link OutputStream}
+ * Named <code>oswrite</code> so does not clash with {@link #write(KeyValue, DataOutput)}
+ * @param kv
+ * @param out
+ * @return Length written on stream
+ * @throws IOException
+ * @see #create(DataInput) for the inverse function
+ * @see #write(KeyValue, DataOutput)
+ */
+ public static long oswrite(final KeyValue kv, final OutputStream out) throws IOException {
+ int length = kv.getLength();
+ // This does same as DataOuput#writeInt (big-endian, etc.)
+ out.write(Bytes.toBytes(length));
+ out.write(kv.getBuffer(), kv.getOffset(), length);
+ return length + Bytes.SIZEOF_INT;
+ }
+
+ /**
* Compare key portion of a {@link KeyValue} for keys in <code>-ROOT-<code>
* table.
*/
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTestUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTestUtil.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTestUtil.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTestUtil.java Sat Feb 23 22:38:58 2013
@@ -26,7 +26,7 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.IterableUtils;
import org.apache.hadoop.hbase.util.Strings;
-import org.apache.hbase.cell.CellComparator;
+import org.apache.hbase.CellComparator;
import com.google.common.collect.Lists;
@@ -63,10 +63,10 @@ public class KeyValueTestUtil {
public static ByteBuffer toByteBufferAndRewind(final Iterable<? extends KeyValue> kvs,
boolean includeMemstoreTS) {
- int totalBytes = KeyValueTool.totalLengthWithMvccVersion(kvs, includeMemstoreTS);
+ int totalBytes = KeyValueUtil.totalLengthWithMvccVersion(kvs, includeMemstoreTS);
ByteBuffer bb = ByteBuffer.allocate(totalBytes);
for (KeyValue kv : IterableUtils.nullSafe(kvs)) {
- KeyValueTool.appendToByteBuffer(bb, kv, includeMemstoreTS);
+ KeyValueUtil.appendToByteBuffer(bb, kv, includeMemstoreTS);
}
bb.rewind();
return bb;
@@ -99,7 +99,7 @@ public class KeyValueTestUtil {
List<KeyValue> kvs = Lists.newArrayList();
KeyValue kv = null;
while (true) {
- kv = KeyValueTool.nextShallowCopy(bb, includesMemstoreTS);
+ kv = KeyValueUtil.nextShallowCopy(bb, includesMemstoreTS);
if (kv == null) {
break;
}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueUtil.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueUtil.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueUtil.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,207 @@
+/*
+ * 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.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.ByteRange;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.IterableUtils;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hbase.Cell;
+import org.apache.hbase.CellUtil;
+
+/**
+ * static convenience methods for dealing with KeyValues and collections of KeyValues
+ */
+@InterfaceAudience.Private
+public class KeyValueUtil {
+
+ /**************** length *********************/
+
+ public static int length(final Cell cell) {
+ return (int)KeyValue.getKeyValueDataStructureSize(cell.getRowLength(), cell.getFamilyLength(),
+ cell.getQualifierLength(), cell.getValueLength());
+ }
+
+ protected static int keyLength(final Cell cell) {
+ return (int)KeyValue.getKeyDataStructureSize(cell.getRowLength(), cell.getFamilyLength(),
+ cell.getQualifierLength());
+ }
+
+ 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<? extends KeyValue> kvs,
+ final boolean includeMvccVersion) {
+ int length = 0;
+ for (KeyValue kv : IterableUtils.nullSafe(kvs)) {
+ length += lengthWithMvccVersion(kv, includeMvccVersion);
+ }
+ return length;
+ }
+
+
+ /**************** copy key only *********************/
+
+ public static KeyValue copyToNewKeyValue(final Cell cell) {
+ KeyValue kvCell = new KeyValue(copyToNewByteArray(cell));
+ kvCell.setMvccVersion(cell.getMvccVersion());
+ return kvCell;
+ }
+
+ public static ByteBuffer copyKeyToNewByteBuffer(final Cell cell) {
+ byte[] bytes = new byte[keyLength(cell)];
+ appendKeyToByteArrayWithoutValue(cell, bytes, 0);
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ buffer.position(buffer.limit());//make it look as if each field were appended
+ return buffer;
+ }
+
+ public static byte[] copyToNewByteArray(final Cell cell) {
+ int v1Length = length(cell);
+ byte[] backingBytes = new byte[v1Length];
+ appendToByteArray(cell, backingBytes, 0);
+ return backingBytes;
+ }
+
+ protected static int appendKeyToByteArrayWithoutValue(final Cell cell, final byte[] output,
+ final int offset) {
+ int nextOffset = offset;
+ nextOffset = Bytes.putShort(output, nextOffset, cell.getRowLength());
+ nextOffset = CellUtil.copyRowTo(cell, output, nextOffset);
+ nextOffset = Bytes.putByte(output, nextOffset, cell.getFamilyLength());
+ nextOffset = CellUtil.copyFamilyTo(cell, output, nextOffset);
+ nextOffset = CellUtil.copyQualifierTo(cell, output, nextOffset);
+ nextOffset = Bytes.putLong(output, nextOffset, cell.getTimestamp());
+ nextOffset = Bytes.putByte(output, nextOffset, cell.getTypeByte());
+ return nextOffset;
+ }
+
+
+ /**************** copy key and value *********************/
+
+ public static int appendToByteArray(final Cell cell, final byte[] output, final int offset) {
+ int pos = offset;
+ pos = Bytes.putInt(output, pos, keyLength(cell));
+ pos = Bytes.putInt(output, pos, cell.getValueLength());
+ pos = appendKeyToByteArrayWithoutValue(cell, output, pos);
+ CellUtil.copyValueTo(cell, output, pos);
+ return pos + cell.getValueLength();
+ }
+
+ public static ByteBuffer copyToNewByteBuffer(final Cell cell) {
+ byte[] bytes = new byte[length(cell)];
+ appendToByteArray(cell, bytes, 0);
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ buffer.position(buffer.limit());//make it look as if each field were appended
+ return buffer;
+ }
+
+ public static void appendToByteBuffer(final ByteBuffer bb, final KeyValue kv,
+ final boolean includeMvccVersion) {
+ // keep pushing the limit out. assume enough capacity
+ bb.limit(bb.position() + kv.getLength());
+ bb.put(kv.getBuffer(), kv.getOffset(), kv.getLength());
+ if (includeMvccVersion) {
+ int numMvccVersionBytes = WritableUtils.getVIntSize(kv.getMvccVersion());
+ ByteBufferUtils.extendLimit(bb, numMvccVersionBytes);
+ ByteBufferUtils.writeVLong(bb, kv.getMvccVersion());
+ }
+ }
+
+
+ /**************** iterating *******************************/
+
+ /**
+ * Creates a new KeyValue object positioned in the supplied ByteBuffer and sets the ByteBuffer's
+ * position to the start of the next KeyValue. Does not allocate a new array or copy data.
+ */
+ public static KeyValue nextShallowCopy(final ByteBuffer bb, final boolean includesMvccVersion) {
+ if (bb.isDirect()) {
+ throw new IllegalArgumentException("only supports heap buffers");
+ }
+ if (bb.remaining() < 1) {
+ return null;
+ }
+ int underlyingArrayOffset = bb.arrayOffset() + bb.position();
+ int keyLength = bb.getInt();
+ int valueLength = bb.getInt();
+ int kvLength = KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + keyLength + valueLength;
+ KeyValue keyValue = new KeyValue(bb.array(), underlyingArrayOffset, kvLength);
+ ByteBufferUtils.skip(bb, keyLength + valueLength);
+ if (includesMvccVersion) {
+ long mvccVersion = ByteBufferUtils.readVLong(bb);
+ keyValue.setMvccVersion(mvccVersion);
+ }
+ return keyValue;
+ }
+
+
+ /*************** next/previous **********************************/
+
+ /**
+ * Append single byte 0x00 to the end of the input row key
+ */
+ public static KeyValue createFirstKeyInNextRow(final Cell in){
+ byte[] nextRow = new byte[in.getRowLength() + 1];
+ System.arraycopy(in.getRowArray(), in.getRowOffset(), nextRow, 0, in.getRowLength());
+ nextRow[nextRow.length - 1] = 0;//maybe not necessary
+ return KeyValue.createFirstOnRow(nextRow);
+ }
+
+ /**
+ * Increment the row bytes and clear the other fields
+ */
+ public static KeyValue createFirstKeyInIncrementedRow(final Cell in){
+ byte[] thisRow = new ByteRange(in.getRowArray(), in.getRowOffset(), in.getRowLength())
+ .deepCopyToNewArray();
+ byte[] nextRow = Bytes.unsignedCopyAndIncrement(thisRow);
+ return KeyValue.createFirstOnRow(nextRow);
+ }
+
+ /**
+ * Decrement the timestamp. For tests (currently wasteful)
+ *
+ * Remember timestamps are sorted reverse chronologically.
+ * @param in
+ * @return previous key
+ */
+ public static KeyValue previousKey(final KeyValue in) {
+ return KeyValue.createFirstOnRow(CellUtil.getRowArray(in), CellUtil.getFamilyArray(in),
+ CellUtil.getQualifierArray(in), in.getTimestamp() - 1);
+ }
+
+ /*************** misc **********************************/
+ /**
+ * @param cell
+ * @return <code>cell<code> if it is an instance of {@link KeyValue} else we will return a
+ * new {@link KeyValue} instance made from <code>cell</code>
+ */
+ public static Cell ensureKeyValue(final Cell cell) {
+ return cell instanceof KeyValue? cell: copyToNewKeyValue(cell);
+ }
+}
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ImmutableBytesWritable.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ImmutableBytesWritable.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ImmutableBytesWritable.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ImmutableBytesWritable.java Sat Feb 23 22:38:58 2013
@@ -41,6 +41,9 @@ import org.apache.hadoop.io.WritableComp
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
+@edu.umd.cs.findbugs.annotations.SuppressWarnings(
+ value="EQ_CHECK_FOR_OPERAND_NOT_COMPATIBLE_WITH_THIS",
+ justification="It has been like this forever")
public class ImmutableBytesWritable
implements WritableComparable<ImmutableBytesWritable> {
private byte[] bytes;
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java Sat Feb 23 22:38:58 2013
@@ -97,6 +97,9 @@ public final class Compression {
* risk breaking all existing HFiles out there. Even the ones that are
* not compressed! (They use the NONE algorithm)
*/
+ @edu.umd.cs.findbugs.annotations.SuppressWarnings(
+ value="SE_TRANSIENT_FIELD_NOT_RESTORED",
+ justification="We are not serializing so doesn't apply (not sure why transient though)")
public static enum Algorithm {
LZO("lzo") {
// Use base type to avoid compile-time dependencies.
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java Sat Feb 23 22:38:58 2013
@@ -628,7 +628,9 @@ public class Bytes {
}
/**
- * Convert an int value to a byte array
+ * Convert an int value to a byte array. Big-endian. Same as what DataOutputStream.writeInt
+ * does.
+ *
* @param val value
* @return the byte array
*/
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/JVM.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/JVM.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/JVM.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/JVM.java Sat Feb 23 22:38:58 2013
@@ -127,6 +127,7 @@ public class JVM
return (ofdc != null ? ofdc.longValue () : -1);
}
InputStream in = null;
+ BufferedReader output = null;
try {
//need to get the PID number of the process first
RuntimeMXBean rtmbean = ManagementFactory.getRuntimeMXBean();
@@ -138,22 +139,27 @@ public class JVM
new String[] { "bash", "-c",
"ls /proc/" + pidhost[0] + "/fdinfo | wc -l" });
in = p.getInputStream();
- BufferedReader output = new BufferedReader(
- new InputStreamReader(in));
-
+ output = new BufferedReader(new InputStreamReader(in));
String openFileDesCount;
if ((openFileDesCount = output.readLine()) != null)
return Long.parseLong(openFileDesCount);
} catch (IOException ie) {
- LOG.warn("Not able to get the number of open file descriptors", ie);
- } finally {
- if (in != null){
- try {
- in.close();
- } catch (IOException e) {
- LOG.warn("Not able to close the InputStream", e);
- }
- }
+ LOG.warn("Not able to get the number of open file descriptors", ie);
+ } finally {
+ if (output != null) {
+ try {
+ output.close();
+ } catch (IOException e) {
+ LOG.warn("Not able to close the InputStream", e);
+ }
+ }
+ if (in != null){
+ try {
+ in.close();
+ } catch (IOException e) {
+ LOG.warn("Not able to close the InputStream", e);
+ }
+ }
}
return -1;
}
@@ -165,30 +171,30 @@ public class JVM
* @return max number of file descriptors the operating system can use.
*/
public long getMaxFileDescriptorCount() {
-
Long mfdc;
-
if (!ibmvendor) {
mfdc = runUnixMXBeanMethod("getMaxFileDescriptorCount");
return (mfdc != null ? mfdc.longValue () : -1);
}
InputStream in = null;
+ BufferedReader output = null;
try {
-
//using linux bash commands to retrieve info
- Process p = Runtime.getRuntime().exec(
- new String[] { "bash", "-c",
- "ulimit -n" });
+ Process p = Runtime.getRuntime().exec(new String[] { "bash", "-c", "ulimit -n" });
in = p.getInputStream();
- BufferedReader output = new BufferedReader(
- new InputStreamReader(in));
-
+ output = new BufferedReader(new InputStreamReader(in));
String maxFileDesCount;
- if ((maxFileDesCount = output.readLine()) != null)
- return Long.parseLong(maxFileDesCount);
- } catch (IOException ie) {
- LOG.warn("Not able to get the max number of file descriptors", ie);
+ if ((maxFileDesCount = output.readLine()) != null) return Long.parseLong(maxFileDesCount);
+ } catch (IOException ie) {
+ LOG.warn("Not able to get the max number of file descriptors", ie);
} finally {
+ if (output != null) {
+ try {
+ output.close();
+ } catch (IOException e) {
+ LOG.warn("Not able to close the reader", e);
+ }
+ }
if (in != null){
try {
in.close();
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/LoadTestKVGenerator.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/LoadTestKVGenerator.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/LoadTestKVGenerator.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/LoadTestKVGenerator.java Sat Feb 23 22:38:58 2013
@@ -16,7 +16,6 @@
*/
package org.apache.hadoop.hbase.util.test;
-import java.util.Map;
import java.util.Random;
import org.apache.hadoop.hbase.util.Bytes;
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/RedundantKVGenerator.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/RedundantKVGenerator.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/RedundantKVGenerator.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hadoop/hbase/util/test/RedundantKVGenerator.java Sat Feb 23 22:38:58 2013
@@ -205,7 +205,7 @@ public class RedundantKVGenerator {
randomizer.nextBytes(family);
}
- long baseTimestamp = Math.abs(randomizer.nextLong()) / baseTimestampDivide;
+ long baseTimestamp = Math.abs(randomizer.nextInt()) / baseTimestampDivide;
byte[] value = new byte[valueLength];
Modified: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/Cell.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/Cell.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/Cell.java (original)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/Cell.java Sat Feb 23 22:38:58 2013
@@ -20,7 +20,6 @@ package org.apache.hbase;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hbase.cell.CellTool;
/**
@@ -43,7 +42,7 @@ import org.apache.hbase.cell.CellTool;
* the goal of sorting newer cells first.
* <p/>
* This interface does not include methods that allocate new byte[]'s such as those used in client
- * or debugging code. These should be placed in a sub-interface or the {@link CellTool} class.
+ * or debugging code. These should be placed in a sub-interface or the {@link CellUtil} class.
* <p/>
* Cell implements Comparable<Cell> which is only meaningful when comparing to other keys in the
* same table. It uses CellComparator which does not work on the -ROOT- and .META. tables.
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellComparator.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellComparator.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellComparator.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellComparator.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,198 @@
+/*
+ * 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.hbase;
+
+import java.io.Serializable;
+import java.util.Comparator;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import com.google.common.primitives.Longs;
+
+/**
+ * Compare two HBase cells. Do not use this method comparing <code>-ROOT-</code> or
+ * <code>.META.</code> cells. Cells from these tables need a specialized comparator, one that
+ * takes account of the special formatting of the row where we have commas to delimit table from
+ * regionname, from row. See KeyValue for how it has a special comparator to do .META. cells
+ * and yet another for -ROOT-.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class CellComparator implements Comparator<Cell>, Serializable{
+ private static final long serialVersionUID = -8760041766259623329L;
+
+ @Override
+ public int compare(Cell a, Cell b) {
+ return compareStatic(a, b);
+ }
+
+
+ public static int compareStatic(Cell a, Cell b) {
+ //row
+ int c = Bytes.compareTo(
+ a.getRowArray(), a.getRowOffset(), a.getRowLength(),
+ b.getRowArray(), b.getRowOffset(), b.getRowLength());
+ if (c != 0) return c;
+
+ //family
+ c = Bytes.compareTo(
+ a.getFamilyArray(), a.getFamilyOffset(), a.getFamilyLength(),
+ b.getFamilyArray(), b.getFamilyOffset(), b.getFamilyLength());
+ if (c != 0) return c;
+
+ //qualifier
+ c = Bytes.compareTo(
+ a.getQualifierArray(), a.getQualifierOffset(), a.getQualifierLength(),
+ b.getQualifierArray(), b.getQualifierOffset(), b.getQualifierLength());
+ if (c != 0) return c;
+
+ //timestamp: later sorts first
+ c = -Longs.compare(a.getTimestamp(), b.getTimestamp());
+ if (c != 0) return c;
+
+ //type
+ c = (0xff & a.getTypeByte()) - (0xff & b.getTypeByte());
+ if (c != 0) return c;
+
+ //mvccVersion: later sorts first
+ return -Longs.compare(a.getMvccVersion(), b.getMvccVersion());
+ }
+
+
+ /**************** equals ****************************/
+
+ public static boolean equals(Cell a, Cell b){
+ return equalsRow(a, b)
+ && equalsFamily(a, b)
+ && equalsQualifier(a, b)
+ && equalsTimestamp(a, b)
+ && equalsType(a, b);
+ }
+
+ public static boolean equalsRow(Cell a, Cell b){
+ return Bytes.equals(
+ a.getRowArray(), a.getRowOffset(), a.getRowLength(),
+ b.getRowArray(), b.getRowOffset(), b.getRowLength());
+ }
+
+ public static boolean equalsFamily(Cell a, Cell b){
+ return Bytes.equals(
+ a.getFamilyArray(), a.getFamilyOffset(), a.getFamilyLength(),
+ b.getFamilyArray(), b.getFamilyOffset(), b.getFamilyLength());
+ }
+
+ public static boolean equalsQualifier(Cell a, Cell b){
+ return Bytes.equals(
+ a.getQualifierArray(), a.getQualifierOffset(), a.getQualifierLength(),
+ b.getQualifierArray(), b.getQualifierOffset(), b.getQualifierLength());
+ }
+
+ public static boolean equalsTimestamp(Cell a, Cell b){
+ return a.getTimestamp() == b.getTimestamp();
+ }
+
+ public static boolean equalsType(Cell a, Cell b){
+ return a.getTypeByte() == b.getTypeByte();
+ }
+
+
+ /********************* hashCode ************************/
+
+ /**
+ * Returns a hash code that is always the same for two Cells having a matching equals(..) result.
+ * Currently does not guard against nulls, but it could if necessary.
+ */
+ public static int hashCode(Cell cell){
+ if (cell == null) {// return 0 for empty Cell
+ return 0;
+ }
+
+ //pre-calculate the 3 hashes made of byte ranges
+ int rowHash = Bytes.hashCode(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
+ int familyHash =
+ Bytes.hashCode(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
+ int qualifierHash = Bytes.hashCode(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength());
+
+ //combine the 6 sub-hashes
+ int hash = 31 * rowHash + familyHash;
+ hash = 31 * hash + qualifierHash;
+ hash = 31 * hash + (int)cell.getTimestamp();
+ hash = 31 * hash + cell.getTypeByte();
+ hash = 31 * hash + (int)cell.getMvccVersion();
+ return hash;
+ }
+
+
+ /******************** lengths *************************/
+
+ public static boolean areKeyLengthsEqual(Cell a, Cell b) {
+ return a.getRowLength() == b.getRowLength()
+ && a.getFamilyLength() == b.getFamilyLength()
+ && a.getQualifierLength() == b.getQualifierLength();
+ }
+
+ public static boolean areRowLengthsEqual(Cell a, Cell b) {
+ return a.getRowLength() == b.getRowLength();
+ }
+
+
+ /***************** special cases ****************************/
+
+ /**
+ * special case for KeyValue.equals
+ */
+ private static int compareStaticIgnoreMvccVersion(Cell a, Cell b) {
+ //row
+ int c = Bytes.compareTo(
+ a.getRowArray(), a.getRowOffset(), a.getRowLength(),
+ b.getRowArray(), b.getRowOffset(), b.getRowLength());
+ if (c != 0) return c;
+
+ //family
+ c = Bytes.compareTo(
+ a.getFamilyArray(), a.getFamilyOffset(), a.getFamilyLength(),
+ b.getFamilyArray(), b.getFamilyOffset(), b.getFamilyLength());
+ if (c != 0) return c;
+
+ //qualifier
+ c = Bytes.compareTo(
+ a.getQualifierArray(), a.getQualifierOffset(), a.getQualifierLength(),
+ b.getQualifierArray(), b.getQualifierOffset(), b.getQualifierLength());
+ if (c != 0) return c;
+
+ //timestamp: later sorts first
+ c = -Longs.compare(a.getTimestamp(), b.getTimestamp());
+ if (c != 0) return c;
+
+ //type
+ c = (0xff & a.getTypeByte()) - (0xff & b.getTypeByte());
+ return c;
+ }
+
+ /**
+ * special case for KeyValue.equals
+ */
+ public static boolean equalsIgnoreMvccVersion(Cell a, Cell b){
+ return 0 == compareStaticIgnoreMvccVersion(a, b);
+ }
+
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScannable.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScannable.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScannable.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScannable.java Sat Feb 23 22:38:58 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.hbase;
+
+/**
+ * Implementer can return a CellScanner over its Cell content.
+ * Class name is ugly but mimicing java.util.Iterable only we are about the dumber
+ * CellScanner rather than say Iterator<Cell>. See CellScanner class comment for why we go
+ * dumber than java.util.Iterator.
+ */
+public interface CellScannable {
+ /**
+ * @return A CellScanner over the contained {@link Cell}s
+ */
+ CellScanner cellScanner();
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScanner.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScanner.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScanner.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellScanner.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,62 @@
+/**
+ * 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.hbase;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hbase.Cell;
+
+/**
+ * An interface for iterating through a sequence of cells. Similar to Java's Iterator, but without
+ * the hasNext() or remove() methods. The hasNext() method is problematic because it may require
+ * actually loading the next object, which in turn requires storing the previous object somewhere.
+ *
+ * <p>The core data block decoder should be as fast as possible, so we push the complexity and
+ * performance expense of concurrently tracking multiple cells to layers above the CellScanner.
+ * <p>
+ * The {@link #current()} method will return a reference to a Cell implementation. This reference
+ * may or may not point to a reusable cell implementation, so users of the CellScanner should not,
+ * for example, accumulate a List of Cells. All of the references may point to the same object,
+ * which would be the latest state of the underlying Cell. In short, the Cell is mutable.
+ * <p/>
+ * Typical usage:
+ *
+ * <pre>
+ * while (scanner.next()) {
+ * Cell cell = scanner.get();
+ * // do something
+ * }
+ * </pre>
+ * <p>Often used reading {@link org.apache.hbase.Cell}s written by
+ * {@link org.apache.hbase.io.CellOutputStream}.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public interface CellScanner {
+ /**
+ * @return the current Cell which may be mutable
+ */
+ Cell current();
+
+ /**
+ * Advance the scanner 1 cell.
+ * @return true if the next cell is found and {@link #current()} will return a valid Cell
+ */
+ boolean advance();
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellUtil.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellUtil.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/CellUtil.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,243 @@
+/*
+ * 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.hbase;
+
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.util.ByteRange;
+import org.apache.hadoop.hbase.KeyValue;
+
+/**
+ * Utility methods helpful slinging {@link Cell} instances.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public final class CellUtil {
+
+ /******************* ByteRange *******************************/
+
+ public static ByteRange fillRowRange(Cell cell, ByteRange range) {
+ return range.set(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
+ }
+
+ public static ByteRange fillFamilyRange(Cell cell, ByteRange range) {
+ return range.set(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
+ }
+
+ public static ByteRange fillQualifierRange(Cell cell, ByteRange range) {
+ return range.set(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength());
+ }
+
+
+ /***************** get individual arrays for tests ************/
+
+ public static byte[] getRowArray(Cell cell){
+ byte[] output = new byte[cell.getRowLength()];
+ copyRowTo(cell, output, 0);
+ return output;
+ }
+
+ public static byte[] getFamilyArray(Cell cell){
+ byte[] output = new byte[cell.getFamilyLength()];
+ copyFamilyTo(cell, output, 0);
+ return output;
+ }
+
+ public static byte[] getQualifierArray(Cell cell){
+ byte[] output = new byte[cell.getQualifierLength()];
+ copyQualifierTo(cell, output, 0);
+ return output;
+ }
+
+ public static byte[] getValueArray(Cell cell){
+ byte[] output = new byte[cell.getValueLength()];
+ copyValueTo(cell, output, 0);
+ return output;
+ }
+
+
+ /******************** copyTo **********************************/
+
+ public static int copyRowTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getRowArray(), cell.getRowOffset(), destination, destinationOffset,
+ cell.getRowLength());
+ return destinationOffset + cell.getRowLength();
+ }
+
+ public static int copyFamilyTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getFamilyArray(), cell.getFamilyOffset(), destination, destinationOffset,
+ cell.getFamilyLength());
+ return destinationOffset + cell.getFamilyLength();
+ }
+
+ public static int copyQualifierTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getQualifierArray(), cell.getQualifierOffset(), destination,
+ destinationOffset, cell.getQualifierLength());
+ return destinationOffset + cell.getQualifierLength();
+ }
+
+ public static int copyValueTo(Cell cell, byte[] destination, int destinationOffset) {
+ System.arraycopy(cell.getValueArray(), cell.getValueOffset(), destination, destinationOffset,
+ cell.getValueLength());
+ return destinationOffset + cell.getValueLength();
+ }
+
+
+ /********************* misc *************************************/
+
+ public static byte getRowByte(Cell cell, int index) {
+ return cell.getRowArray()[cell.getRowOffset() + index];
+ }
+
+ public static ByteBuffer getValueBufferShallowCopy(Cell cell) {
+ ByteBuffer buffer = ByteBuffer.wrap(cell.getValueArray(), cell.getValueOffset(),
+ cell.getValueLength());
+// buffer.position(buffer.limit());//make it look as if value was appended
+ return buffer;
+ }
+
+ public static Cell createCell(final byte [] row, final byte [] family, final byte [] qualifier,
+ final long timestamp, final byte type, final byte [] value) {
+ // I need a Cell Factory here. Using KeyValue for now. TODO.
+ // TODO: Make a new Cell implementation that just carries these
+ // byte arrays.
+ return new KeyValue(row, family, qualifier, timestamp,
+ KeyValue.Type.codeToType(type), value);
+ }
+
+ /**
+ * @param cellScannerables
+ * @return CellScanner interface over <code>cellIterables</code>
+ */
+ public static CellScanner createCellScanner(final List<CellScannable> cellScannerables) {
+ return new CellScanner() {
+ private final Iterator<CellScannable> iterator = cellScannerables.iterator();
+ private CellScanner cellScanner = null;
+
+ @Override
+ public Cell current() {
+ return this.cellScanner != null? this.cellScanner.current(): null;
+ }
+
+ @Override
+ public boolean advance() {
+ if (this.cellScanner == null) {
+ if (!this.iterator.hasNext()) return false;
+ this.cellScanner = this.iterator.next().cellScanner();
+ }
+ if (this.cellScanner.advance()) return true;
+ this.cellScanner = null;
+ return advance();
+ }
+ };
+ }
+
+ /**
+ * @param cellIterable
+ * @return CellScanner interface over <code>cellIterable</code>
+ */
+ public static CellScanner createCellScanner(final Iterable<Cell> cellIterable) {
+ return createCellScanner(cellIterable.iterator());
+ }
+
+ /**
+ * @param cells
+ * @return CellScanner interface over <code>cellIterable</code>
+ */
+ public static CellScanner createCellScanner(final Iterator<Cell> cells) {
+ return new CellScanner() {
+ private final Iterator<Cell> iterator = cells;
+ private Cell current = null;
+
+ @Override
+ public Cell current() {
+ return this.current;
+ }
+
+ @Override
+ public boolean advance() {
+ boolean hasNext = this.iterator.hasNext();
+ this.current = hasNext? this.iterator.next(): null;
+ return hasNext;
+ }
+ };
+ }
+
+ /**
+ * @param cellArray
+ * @return CellScanner interface over <code>cellArray</code>
+ */
+ public static CellScanner createCellScanner(final Cell[] cellArray) {
+ return new CellScanner() {
+ private final Cell [] cells = cellArray;
+ private int index = -1;
+
+ @Override
+ public Cell current() {
+ return (index < 0)? null: this.cells[index];
+ }
+
+ @Override
+ public boolean advance() {
+ return ++index < this.cells.length;
+ }
+ };
+ }
+
+ /**
+ * Flatten the map of cells out under the CellScanner
+ * @param map Map of Cell Lists; for example, the map of families to Cells that is used
+ * inside Put, etc., keeping Cells organized by family.
+ * @return CellScanner interface over <code>cellIterable</code>
+ */
+ public static CellScanner createCellScanner(final NavigableMap<byte [], List<Cell>> map) {
+ return new CellScanner() {
+ private final Iterator<Entry<byte[], List<Cell>>> entries = map.entrySet().iterator();
+ private Iterator<Cell> currentIterator = null;
+ private Cell currentCell;
+
+ @Override
+ public Cell current() {
+ return this.currentCell;
+ }
+
+ @Override
+ public boolean advance() {
+ if (this.currentIterator == null) {
+ if (!this.entries.hasNext()) return false;
+ this.currentIterator = this.entries.next().getValue().iterator();
+ }
+ if (this.currentIterator.hasNext()) {
+ this.currentCell = this.currentIterator.next();
+ return true;
+ }
+ this.currentCell = null;
+ this.currentIterator = null;
+ return advance();
+ }
+ };
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseDecoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseDecoder.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseDecoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseDecoder.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,59 @@
+/**
+ * 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.hbase.codec;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hbase.Cell;
+
+abstract class BaseDecoder implements Codec.Decoder {
+ final InputStream in;
+ private boolean hasNext = true;
+ private Cell current = null;
+
+ BaseDecoder(final InputStream in) {
+ this.in = in;
+ }
+
+ @Override
+ public boolean advance() {
+ if (!this.hasNext) return this.hasNext;
+ try {
+ if (this.in.available() <= 0) {
+ this.hasNext = false;
+ return this.hasNext;
+ }
+ this.current = parseCell();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return this.hasNext;
+ }
+
+ /**
+ * @return extract a Cell
+ * @throws IOException
+ */
+ abstract Cell parseCell() throws IOException;
+
+ @Override
+ public Cell current() {
+ return this.current;
+ }
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseEncoder.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseEncoder.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseEncoder.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/BaseEncoder.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,51 @@
+/**
+ * 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.hbase.codec;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.hbase.Cell;
+
+abstract class BaseEncoder implements Codec.Encoder {
+ protected final OutputStream out;
+ // This encoder is 'done' once flush has been called.
+ protected boolean flushed = false;
+
+ public BaseEncoder(final OutputStream out) {
+ this.out = out;
+ }
+
+ @Override
+ public abstract void write(Cell cell) throws IOException;
+
+ void checkFlushed() throws CodecException {
+ if (this.flushed) throw new CodecException("Flushed; done");
+ }
+
+ @Override
+ public void flush() throws IOException {
+ if (this.flushed) return;
+ this.flushed = true;
+ try {
+ this.out.flush();
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CellCodec.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CellCodec.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CellCodec.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CellCodec.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,115 @@
+/**
+ * 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.hbase.codec;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hbase.Cell;
+import org.apache.hbase.CellUtil;
+
+/**
+ * Basic Cell codec that just writes out all the individual elements of a Cell. Uses ints
+ * delimiting all lengths. Profligate. Needs tune up. Does not write the mvcc stamp.
+ * Use a different codec if you want that in the stream.
+ */
+public class CellCodec implements Codec {
+ static class CellEncoder extends BaseEncoder {
+ CellEncoder(final OutputStream out) {
+ super(out);
+ }
+
+ @Override
+ public void write(Cell cell) throws IOException {
+ checkFlushed();
+ try {
+ // Row
+ write(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
+ // Column family
+ write(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
+ // Qualifier
+ write(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
+ // Version
+ this.out.write(Bytes.toBytes(cell.getTimestamp()));
+ // Type
+ this.out.write(cell.getTypeByte());
+ // Value
+ write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ }
+
+ /**
+ * Write int length followed by array bytes.
+ * @param bytes
+ * @param offset
+ * @param length
+ * @throws IOException
+ */
+ private void write(final byte [] bytes, final int offset, final int length)
+ throws IOException {
+ this.out.write(Bytes.toBytes(length));
+ this.out.write(bytes, offset, length);
+ }
+ }
+
+ static class CellDecoder extends BaseDecoder {
+ public CellDecoder(final InputStream in) {
+ super(in);
+ }
+
+ Cell parseCell() throws IOException {
+ byte [] row = readByteArray(this.in);
+ byte [] family = readByteArray(in);
+ byte [] qualifier = readByteArray(in);
+ byte [] longArray = new byte[Bytes.SIZEOF_LONG];
+ IOUtils.readFully(this.in, longArray);
+ long timestamp = Bytes.toLong(longArray);
+ byte type = (byte) this.in.read();
+ byte [] value = readByteArray(in);
+ return CellUtil.createCell(row, family, qualifier, timestamp, type, value);
+ }
+
+ /**
+ * @return Byte array read from the stream.
+ * @throws IOException
+ */
+ private byte [] readByteArray(final InputStream in) throws IOException {
+ byte [] intArray = new byte[Bytes.SIZEOF_INT];
+ IOUtils.readFully(in, intArray);
+ int length = Bytes.toInt(intArray);
+ byte [] bytes = new byte [length];
+ IOUtils.readFully(in, bytes);
+ return bytes;
+ }
+ }
+
+ @Override
+ public Decoder getDecoder(InputStream is) {
+ return new CellDecoder(is);
+ }
+
+ @Override
+ public Encoder getEncoder(OutputStream os) {
+ return new CellEncoder(os);
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/Codec.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/Codec.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/Codec.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/Codec.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,51 @@
+/**
+ * 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.hbase.codec;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoder;
+import org.apache.hbase.CellScanner;
+import org.apache.hbase.io.CellOutputStream;
+
+/**
+ * Encoder/Decoder for Cell.
+ *
+ * <p>Like {@link DataBlockEncoder} only Cell-based rather than KeyValue version 1 based
+ * and without presuming an hfile context. Intent is an Interface that will work for hfile and
+ * rpc.
+ */
+public interface Codec {
+ // TODO: interfacing with {@link DataBlockEncoder}
+ /**
+ * Call flush when done. Some encoders may not put anything on the stream until flush is called.
+ * On flush, let go of any resources used by the encoder.
+ */
+ public interface Encoder extends CellOutputStream {}
+
+ /**
+ * Implementations should implicitly clean up any resources allocated when the
+ * Decoder/CellScanner runs off the end of the cell block. Do this rather than require the user
+ * call close explicitly.
+ */
+ public interface Decoder extends CellScanner {};
+
+ Decoder getDecoder(InputStream is);
+ Encoder getEncoder(OutputStream os);
+}
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,39 @@
+/**
+ * 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.hbase.codec;
+
+import java.io.IOException;
+
+public class CodecException extends IOException {
+ private static final long serialVersionUID = -2850095011686914405L;
+
+ public CodecException() {
+ }
+
+ public CodecException(String message) {
+ super(message);
+ }
+
+ public CodecException(Throwable t) {
+ super(t);
+ }
+
+ public CodecException(String message, Throwable t) {
+ super(message, t);
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/KeyValueCodec.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/KeyValueCodec.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/KeyValueCodec.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/codec/KeyValueCodec.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,86 @@
+/**
+ * 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.hbase.codec;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueUtil;
+import org.apache.hbase.Cell;
+
+/**
+ * Codec that does KeyValue version 1 serialization.
+ *
+ * <p>Encodes by casting Cell to KeyValue and writing out the backing array with a length prefix.
+ * This is how KVs were serialized in Puts, Deletes and Results pre-0.96. Its what would
+ * happen if you called the Writable#write KeyValue implementation. This encoder will fail
+ * if the passed Cell is not an old-school pre-0.96 KeyValue. Does not copy bytes writing.
+ * It just writes them direct to the passed stream.
+ *
+ * <p>If you wrote two KeyValues to this encoder, it would look like this in the stream:
+ * <pre>
+ * length-of-KeyValue1 // A java int with the length of KeyValue1 backing array
+ * KeyValue1 backing array filled with a KeyValue serialized in its particular format
+ * length-of-KeyValue2
+ * KeyValue2 backing array
+ * </pre>
+ */
+public class KeyValueCodec implements Codec {
+ static class KeyValueEncoder extends BaseEncoder {
+ KeyValueEncoder(final OutputStream out) {
+ super(out);
+ }
+
+ @Override
+ public void write(Cell cell) throws IOException {
+ checkFlushed();
+ // This is crass and will not work when KV changes. Also if passed a non-kv Cell, it will
+ // make expensive copy.
+ try {
+ KeyValue.oswrite((KeyValue)KeyValueUtil.ensureKeyValue(cell), this.out);
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ }
+ }
+
+ static class KeyValueDecoder extends BaseDecoder {
+ KeyValueDecoder(final InputStream in) {
+ super(in);
+ }
+
+ Cell parseCell() throws IOException {
+ return KeyValue.iscreate(in);
+ }
+ }
+
+ /**
+ * Implementation depends on {@link InputStream#available()}
+ */
+ @Override
+ public Decoder getDecoder(final InputStream is) {
+ return new KeyValueDecoder(is);
+ }
+
+ @Override
+ public Encoder getEncoder(OutputStream os) {
+ return new KeyValueEncoder(os);
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java (added)
+++ hbase/trunk/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,55 @@
+/*
+ * 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.hbase.io;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hbase.Cell;
+import org.apache.hbase.CellScanner;
+
+/**
+ * Accepts a stream of Cells. This can be used to build a block of cells during compactions
+ * and flushes, or to build a byte[] to send to the client. This could be backed by a
+ * List<KeyValue>, but more efficient implementations will append results to a
+ * byte[] to eliminate overhead, and possibly encode the cells further.
+ * <p>To read Cells, use {@link CellScanner}
+ * @see CellScanner
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface CellOutputStream {
+ /**
+ * Implementation must copy the entire state of the Cell. If the written Cell is modified
+ * immediately after the write method returns, the modifications must have absolutely no effect
+ * on the copy of the Cell that was added in the write.
+ * @param cell Cell to write out
+ * @throws IOException
+ */
+ void write(Cell cell) throws IOException;
+
+ /**
+ * Let the implementation decide what to do. Usually means writing accumulated data into a
+ * byte[] that can then be read from the implementation to be sent to disk, put in the block
+ * cache, or sent over the network.
+ * @throws IOException
+ */
+ void flush() throws IOException;
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/test/java/org/apache/hbase/TestCellUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/test/java/org/apache/hbase/TestCellUtil.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/test/java/org/apache/hbase/TestCellUtil.java (added)
+++ hbase/trunk/hbase-common/src/test/java/org/apache/hbase/TestCellUtil.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,72 @@
+/**
+ * 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.hbase;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Test;
+
+public class TestCellUtil {
+ @Test
+ public void testCreateCellScannerCellList() {
+ final int count = 3;
+ Cell [] cs = getCells(count, Bytes.toBytes(0));
+ List<Cell> cells = Arrays.asList(cs);
+ CellScanner scanner = CellUtil.createCellScanner(cells);
+ int i = 0;
+ while (scanner.advance()) {
+ i++;
+ }
+ assertEquals(count, i);
+ }
+
+ @Test
+ public void testCreateCellScannerFamilyMap() {
+ final int count = 3;
+ final NavigableMap<byte [], List<Cell>> map =
+ new TreeMap<byte [], List<Cell>>(Bytes.BYTES_COMPARATOR);
+ for (int i = 0; i < count; i++) {
+ byte [] key = Bytes.toBytes(i);
+ Cell [] cs = getCells(count, key);
+ map.put(key, Arrays.asList(cs));
+ }
+ CellScanner scanner = CellUtil.createCellScanner(map);
+ int i = 0;
+ while (scanner.advance()) {
+ i++;
+ }
+ assertEquals(count * count, i);
+ }
+
+ static Cell [] getCells(final int howMany, final byte [] family) {
+ Cell [] cells = new Cell[howMany];
+ for (int i = 0; i < howMany; i++) {
+ byte [] index = Bytes.toBytes(i);
+ KeyValue kv = new KeyValue(index, family, index, index);
+ cells[i] = kv;
+ }
+ return cells;
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java (added)
+++ hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,122 @@
+/**
+ * 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.hbase.codec;
+
+import static org.junit.Assert.*;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hbase.Cell;
+import org.apache.hbase.CellComparator;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.google.common.io.CountingInputStream;
+import com.google.common.io.CountingOutputStream;
+
+@Category(SmallTests.class)
+public class TestCellCodec {
+
+ @Test
+ public void testEmptyWorks() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ CountingOutputStream cos = new CountingOutputStream(baos);
+ DataOutputStream dos = new DataOutputStream(cos);
+ Codec codec = new CellCodec();
+ Codec.Encoder encoder = codec.getEncoder(dos);
+ encoder.flush();
+ dos.close();
+ long offset = cos.getCount();
+ assertEquals(0, offset);
+ CountingInputStream cis =
+ new CountingInputStream(new ByteArrayInputStream(baos.toByteArray()));
+ DataInputStream dis = new DataInputStream(cis);
+ Codec.Decoder decoder = codec.getDecoder(dis);
+ assertFalse(decoder.advance());
+ dis.close();
+ assertEquals(0, cis.getCount());
+ }
+
+ @Test
+ public void testOne() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ CountingOutputStream cos = new CountingOutputStream(baos);
+ DataOutputStream dos = new DataOutputStream(cos);
+ Codec codec = new CellCodec();
+ Codec.Encoder encoder = codec.getEncoder(dos);
+ final KeyValue kv =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("v"));
+ encoder.write(kv);
+ encoder.flush();
+ dos.close();
+ long offset = cos.getCount();
+ CountingInputStream cis =
+ new CountingInputStream(new ByteArrayInputStream(baos.toByteArray()));
+ DataInputStream dis = new DataInputStream(cis);
+ Codec.Decoder decoder = codec.getDecoder(dis);
+ assertTrue(decoder.advance()); // First read should pull in the KV
+ // Second read should trip over the end-of-stream marker and return false
+ assertFalse(decoder.advance());
+ dis.close();
+ assertEquals(offset, cis.getCount());
+ }
+
+ @Test
+ public void testThree() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ CountingOutputStream cos = new CountingOutputStream(baos);
+ DataOutputStream dos = new DataOutputStream(cos);
+ Codec codec = new CellCodec();
+ Codec.Encoder encoder = codec.getEncoder(dos);
+ final KeyValue kv1 =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("1"), Bytes.toBytes("1"));
+ final KeyValue kv2 =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("2"), Bytes.toBytes("2"));
+ final KeyValue kv3 =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("3"), Bytes.toBytes("3"));
+ encoder.write(kv1);
+ encoder.write(kv2);
+ encoder.write(kv3);
+ encoder.flush();
+ dos.close();
+ long offset = cos.getCount();
+ CountingInputStream cis =
+ new CountingInputStream(new ByteArrayInputStream(baos.toByteArray()));
+ DataInputStream dis = new DataInputStream(cis);
+ Codec.Decoder decoder = codec.getDecoder(dis);
+ assertTrue(decoder.advance());
+ Cell c = decoder.current();
+ assertTrue(CellComparator.equals(c, kv1));
+ assertTrue(decoder.advance());
+ c = decoder.current();
+ assertTrue(CellComparator.equals(c, kv2));
+ assertTrue(decoder.advance());
+ c = decoder.current();
+ assertTrue(CellComparator.equals(c, kv3));
+ assertFalse(decoder.advance());
+ dis.close();
+ assertEquals(offset, cis.getCount());
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestKeyValueCodec.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestKeyValueCodec.java?rev=1449420&view=auto
==============================================================================
--- hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestKeyValueCodec.java (added)
+++ hbase/trunk/hbase-common/src/test/java/org/apache/hbase/codec/TestKeyValueCodec.java Sat Feb 23 22:38:58 2013
@@ -0,0 +1,125 @@
+/**
+ * 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.hbase.codec;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.google.common.io.CountingInputStream;
+import com.google.common.io.CountingOutputStream;
+
+@Category(SmallTests.class)
+public class TestKeyValueCodec {
+ @Test
+ public void testEmptyWorks() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ CountingOutputStream cos = new CountingOutputStream(baos);
+ DataOutputStream dos = new DataOutputStream(cos);
+ KeyValueCodec kvc = new KeyValueCodec();
+ Codec.Encoder encoder = kvc.getEncoder(dos);
+ encoder.flush();
+ dos.close();
+ long offset = cos.getCount();
+ assertEquals(0, offset);
+ CountingInputStream cis =
+ new CountingInputStream(new ByteArrayInputStream(baos.toByteArray()));
+ DataInputStream dis = new DataInputStream(cis);
+ Codec.Decoder decoder = kvc.getDecoder(dis);
+ assertFalse(decoder.advance());
+ dis.close();
+ assertEquals(0, cis.getCount());
+ }
+
+ @Test
+ public void testOne() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ CountingOutputStream cos = new CountingOutputStream(baos);
+ DataOutputStream dos = new DataOutputStream(cos);
+ KeyValueCodec kvc = new KeyValueCodec();
+ Codec.Encoder encoder = kvc.getEncoder(dos);
+ final KeyValue kv =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("v"));
+ final long length = kv.getLength() + Bytes.SIZEOF_INT;
+ encoder.write(kv);
+ encoder.flush();
+ dos.close();
+ long offset = cos.getCount();
+ assertEquals(length, offset);
+ CountingInputStream cis =
+ new CountingInputStream(new ByteArrayInputStream(baos.toByteArray()));
+ DataInputStream dis = new DataInputStream(cis);
+ Codec.Decoder decoder = kvc.getDecoder(dis);
+ assertTrue(decoder.advance()); // First read should pull in the KV
+ // Second read should trip over the end-of-stream marker and return false
+ assertFalse(decoder.advance());
+ dis.close();
+ assertEquals(length, cis.getCount());
+ }
+
+ @Test
+ public void testThree() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ CountingOutputStream cos = new CountingOutputStream(baos);
+ DataOutputStream dos = new DataOutputStream(cos);
+ KeyValueCodec kvc = new KeyValueCodec();
+ Codec.Encoder encoder = kvc.getEncoder(dos);
+ final KeyValue kv1 =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("1"), Bytes.toBytes("1"));
+ final KeyValue kv2 =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("2"), Bytes.toBytes("2"));
+ final KeyValue kv3 =
+ new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("3"), Bytes.toBytes("3"));
+ final long length = kv1.getLength() + Bytes.SIZEOF_INT;
+ encoder.write(kv1);
+ encoder.write(kv2);
+ encoder.write(kv3);
+ encoder.flush();
+ dos.close();
+ long offset = cos.getCount();
+ assertEquals(length * 3, offset);
+ CountingInputStream cis =
+ new CountingInputStream(new ByteArrayInputStream(baos.toByteArray()));
+ DataInputStream dis = new DataInputStream(cis);
+ Codec.Decoder decoder = kvc.getDecoder(dis);
+ assertTrue(decoder.advance());
+ KeyValue kv = (KeyValue)decoder.current();
+ assertTrue(kv1.equals(kv));
+ assertTrue(decoder.advance());
+ kv = (KeyValue)decoder.current();
+ assertTrue(kv2.equals(kv));
+ assertTrue(decoder.advance());
+ kv = (KeyValue)decoder.current();
+ assertTrue(kv3.equals(kv));
+ assertFalse(decoder.advance());
+ dis.close();
+ assertEquals((length * 3), cis.getCount());
+ }
+}
\ No newline at end of file
Modified: hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/PrefixTreeCodec.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/PrefixTreeCodec.java?rev=1449420&r1=1449419&r2=1449420&view=diff
==============================================================================
--- hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/PrefixTreeCodec.java (original)
+++ hbase/trunk/hbase-prefix-tree/src/main/java/org/apache/hbase/codec/prefixtree/PrefixTreeCodec.java Sat Feb 23 22:38:58 2013
@@ -28,7 +28,7 @@ import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.KeyComparator;
import org.apache.hadoop.hbase.KeyValue.MetaKeyComparator;
import org.apache.hadoop.hbase.KeyValue.RootKeyComparator;
-import org.apache.hadoop.hbase.KeyValueTool;
+import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoder;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
@@ -97,7 +97,7 @@ public class PrefixTreeCodec implements
try{
KeyValue kv;
- while ((kv = KeyValueTool.nextShallowCopy(rawKeyValues, includesMvccVersion)) != null) {
+ while ((kv = KeyValueUtil.nextShallowCopy(rawKeyValues, includesMvccVersion)) != null) {
builder.write(kv);
}
builder.flush();
@@ -132,13 +132,13 @@ public class PrefixTreeCodec implements
CellSearcher searcher = null;
try {
searcher = DecoderFactory.checkOut(sourceAsBuffer, includesMvccVersion);
- while (searcher.next()) {
- KeyValue currentCell = KeyValueTool.copyToNewKeyValue(searcher.getCurrent());
+ while (searcher.advance()) {
+ KeyValue currentCell = KeyValueUtil.copyToNewKeyValue(searcher.current());
// needs to be modified for DirectByteBuffers. no existing methods to
// write VLongs to byte[]
int offset = result.arrayOffset() + result.position();
- KeyValueTool.appendToByteArray(currentCell, result.array(), offset);
- int keyValueLength = KeyValueTool.length(currentCell);
+ KeyValueUtil.appendToByteArray(currentCell, result.array(), offset);
+ int keyValueLength = KeyValueUtil.length(currentCell);
ByteBufferUtils.skip(result, keyValueLength);
offset += keyValueLength;
if (includesMvccVersion) {
@@ -163,7 +163,7 @@ public class PrefixTreeCodec implements
if (!searcher.positionAtFirstCell()) {
return null;
}
- return KeyValueTool.copyKeyToNewByteBuffer(searcher.getCurrent());
+ return KeyValueUtil.copyKeyToNewByteBuffer(searcher.current());
} finally {
DecoderFactory.checkIn(searcher);
}