in
+ * @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 in 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 {
@@ -2301,7 +2303,8 @@ public class KeyValue implements Cell, HeapSize {
* Create a KeyValue reading length from in
* @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 {
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
index a2f75b9..543b5e4 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValueTool.java
@@ -28,7 +28,7 @@ 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.cell.CellTool;
+import org.apache.hbase.CellTool;
/**
* static convenience methods for dealing with KeyValues and collections of KeyValues
diff --git a/hbase-common/src/main/java/org/apache/hbase/Cell.java b/hbase-common/src/main/java/org/apache/hbase/Cell.java
index 8247bab..3a57e8b 100644
--- a/hbase-common/src/main/java/org/apache/hbase/Cell.java
+++ b/hbase-common/src/main/java/org/apache/hbase/Cell.java
@@ -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;
/**
diff --git a/hbase-common/src/main/java/org/apache/hbase/CellComparator.java b/hbase-common/src/main/java/org/apache/hbase/CellComparator.java
new file mode 100644
index 0000000..f6a3c25
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/CellComparator.java
@@ -0,0 +1,177 @@
+/*
+ * 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 traditional HBase cells.
+ *
+ * Note: This comparator is not valid for -ROOT- and .META. tables.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class CellComparator implements ComparatorTo 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 diff --git a/hbase-common/src/main/java/org/apache/hbase/io/CellScanner.java b/hbase-common/src/main/java/org/apache/hbase/io/CellScanner.java new file mode 100644 index 0000000..4196aac --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/io/CellScanner.java @@ -0,0 +1,71 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hbase.io; + +import java.io.IOException; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hbase.Cell; + +/** + * Another name for this class would be CellInputStream. + * + *
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. + * 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. + *
+ * The get() method will return a reference to a Cell implementation. This reference may + * or may not point to a reusable cell implementation, so users of this class 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. + * + * At a minimum, an implementation will need to be able to advance from one cell to the next in a + * LinkedList fashion. The nextQualifier(), nextFamily(), and nextRow() methods can all be + * implemented by calling next(), however, if the DataBlockEncoding supports random access into + * the block then it may provide smarter versions of these methods. + * + * Typical usage: + * + *
+ * while (scanner.next()) {
+ * Cell cell = scanner.getCurrent();
+ * // do something
+ * }
+ *
+ * @see CellOutputStream
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public interface CellScanner {
+ /**
+ * @return the current Cell which may be mutable. Will be null before the first read
+ * has happened.
+ */
+ Cell getCurrent();
+
+ /**
+ * Advance the scanner 1 cell.
+ * @return true if the next cell is found and getCurrentCell() will return a valid Cell
+ * @throws IOException
+ */
+ boolean read() throws IOException;
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/CellScannerPosition.java b/hbase-common/src/main/java/org/apache/hbase/io/CellScannerPosition.java
new file mode 100644
index 0000000..e10c4a9
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/CellScannerPosition.java
@@ -0,0 +1,68 @@
+/**
+ * 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 org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * An indicator of the state of the scanner after an operation such as nextCell() or positionAt(..).
+ * For example:
+ *
+ * Note: This was not added to suggest that HBase should support client facing reverse Scanners, but
+ * because some {@link CellSearcher} implementations, namely PrefixTree, need a method of backing up
+ * if the positionAt(..) method goes past the requested cell.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public interface ReversibleCellScanner extends CellScanner {
+ /**
+ * Try to position the scanner one Cell before the current position.
+ * @return true if the operation was successful, meaning getCurrentCell() will return a valid
+ * Cell.
+ * false if there were no previous cells, meaning getCurrentCell() will return null.
+ * Scanner position will be {@link CellScannerPosition.BEFORE_FIRST}
+ */
+ boolean previous();
+
+ /**
+ * Try to position the scanner in the row before the current row.
+ * @param endOfRow true for the last cell in the previous row; false for the first cell
+ * @return true if the operation was successful, meaning getCurrentCell() will return a valid
+ * Cell.
+ * false if there were no previous cells, meaning getCurrentCell() will return null.
+ * Scanner position will be {@link CellScannerPosition.BEFORE_FIRST}
+ */
+ boolean previousRow(boolean endOfRow);
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/CellDecoder.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/CellDecoder.java
new file mode 100644
index 0000000..e26e729
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/CellDecoder.java
@@ -0,0 +1,66 @@
+package org.apache.hbase.io.codec;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hbase.Cell;
+
+/**
+ * @see CellEncoder
+ */
+public class CellDecoder implements Decoder {
+ private final DataInputStream in;
+ // If true, this decoder is finished decoding.
+ private boolean finished = false;
+
+ public CellDecoder(final DataInputStream in) {
+ this.in = in;
+ }
+
+ @Override
+ public Cell decode() throws CodecException {
+ Cell cell = null;
+ if (!this.finished) {
+ int length;
+ try {
+ length = WritableUtils.readVInt(this.in);
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ if (length == CellEncoder.END_OF_CELLS) {
+ this.finished = true;
+ } else {
+ // Presume we are on to a cell... read it in.
+ try {
+ byte [] row = new byte [length];
+ this.in.readFully(row, 0, length);
+ byte [] family = readByteArray();
+ byte [] qualifier = readByteArray();
+ long timestamp = this.in.readLong();
+ byte type = this.in.readByte();
+ byte [] value = readByteArray();
+ // I need a Cell Factory here. Using KeyValue for now. TODO.
+ cell = new KeyValue(row, family, qualifier, timestamp,
+ KeyValue.Type.codeToType(type), value);
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ }
+ }
+ return cell;
+ }
+
+ /**
+ * @return Byte array read from the stream.
+ * @throws IOException
+ */
+ private byte [] readByteArray() throws IOException {
+ // TODO: A method like this belongs in CellTool?
+ int length = WritableUtils.readVInt(this.in);
+ byte [] bytes = new byte [length];
+ this.in.readFully(bytes, 0, length);
+ return bytes;
+ }
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/CellEncoder.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/CellEncoder.java
new file mode 100644
index 0000000..317f483
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/CellEncoder.java
@@ -0,0 +1,79 @@
+package org.apache.hbase.io.codec;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hbase.Cell;
+
+/**
+ * Cell encoder that just writes out all the individual elements of a Cell.
+ * Does not write the mvcc stamp. Use a different encoder if you want that
+ * in the stream.
+ */
+public class CellEncoder implements Encoder {
+ // TODO: Replace WritableUtils from hadoop vint w/ our own.
+ // TOOD: I need a CellFactory to make Cells with. Using KeyValue for now.
+
+ // We write out an '0' int as marker that there are no more kvs when you call flush.
+ static final int END_OF_CELLS = 0;
+ // Need to be able to write java types such as int and long so need DataOutput.
+ // Want to stream too so DataOutputStream.
+ private final DataOutputStream out;
+ // This encoder is 'done' once flush has been called because on flush we
+ // write out the END_OF_KEYVALUES marker.
+ private boolean finish = false;
+
+ public CellEncoder(final DataOutputStream out) {
+ this.out = out;
+ }
+
+ @Override
+ public Cell encode(Cell cell) throws CodecException {
+ if (this.finish) throw new CodecException("Finished");
+ // This is crass and will not work when KV changes
+ 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.writeLong(cell.getTimestamp());
+ // Type
+ this.out.writeByte(cell.getTypeByte());
+ // Value
+ write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ return cell;
+ }
+
+ /**
+ * Write vint 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 {
+ WritableUtils.writeVInt(this.out, length);
+ this.out.write(bytes, offset, length);
+ }
+
+ @Override
+ public void finish() throws CodecException {
+ if (this.finish) return;
+ this.finish = true;
+ try {
+ // Write out an vint whose value is zero as end of stream.
+ WritableUtils.writeVInt(this.out, END_OF_CELLS);
+ this.out.flush();
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ }
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/CodecException.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/CodecException.java
new file mode 100644
index 0000000..2d539ee
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/CodecException.java
@@ -0,0 +1,22 @@
+package org.apache.hbase.io.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
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/Decoder.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/Decoder.java
new file mode 100644
index 0000000..d9cf79b
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/Decoder.java
@@ -0,0 +1,15 @@
+package org.apache.hbase.io.codec;
+
+import org.apache.hbase.Cell;
+
+/**
+ * Cell decoder
+ */
+public interface Decoder {
+ /**
+ * Decode next cell.
+ * @return Next decoded cell or null if finished decoding
+ * @throws CodecException
+ */
+ Cell decode() throws CodecException;
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/Encoder.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/Encoder.java
new file mode 100644
index 0000000..e9bef80
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/Encoder.java
@@ -0,0 +1,32 @@
+package org.apache.hbase.io.codec;
+
+import org.apache.hbase.Cell;
+
+/**
+ * Cell encoder.
+ */
+public interface Encoder {
+ // TODO: org.apache.avro.io.Encoder is like this only does java primitive types, not Cells,
+ // but it does do 'fixed' bytes.... could serialize Cell and pass these... ugh..
+ // wouldn't buy us much.
+ // TODO: Do a context for encoder and then another for decoder because they
+ // will have different stuff? Or is that something not in this Interface (I had
+ // it in here and then removed it all)
+ /**
+ * Implementation must copy the entire state of the Cell. If the passed Cell is modified
+ * immediately after the encode method returns, the modifications must have absolutely no effect
+ * on the copy of the Cell that was added to the encoder.
+ * @param cell Cell to encode.
+ * @return The passed cell
+ * @throws CodecException
+ */
+ Cell encode(Cell cell) throws CodecException;
+
+ /**
+ * Finish up the encoding. Add END-OF-ENCODING markers or flush the
+ * stream, etc.
+ * You cannot call {@link #encode(Cell)} after invoking this method.
+ * @throws CodecException
+ */
+ void finish() throws CodecException;
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/OldSchoolKeyValueDecoder.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/OldSchoolKeyValueDecoder.java
new file mode 100644
index 0000000..2eb72dc
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/OldSchoolKeyValueDecoder.java
@@ -0,0 +1,43 @@
+package org.apache.hbase.io.codec;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hbase.Cell;
+
+/**
+ * @see OldSchoolKeyValueEncoder
+ */
+public class OldSchoolKeyValueDecoder implements Decoder {
+ private final DataInputStream in;
+ // If true, this decoder is finished decoding.
+ private boolean finished = false;
+
+ public OldSchoolKeyValueDecoder(final DataInputStream in) {
+ this.in = in;
+ }
+
+ @Override
+ public Cell decode() throws CodecException {
+ Cell cell = null;
+ if (!this.finished) {
+ int length;
+ try {
+ length = this.in.readInt();
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ if (length == OldSchoolKeyValueEncoder.END_OF_KEYVALUES) {
+ this.finished = true;
+ } else {
+ try {
+ cell = KeyValue.create(length, this.in);
+ } catch (IOException e) {
+ throw new CodecException(e);
+ }
+ }
+ }
+ return cell;
+ }
+}
\ No newline at end of file
diff --git a/hbase-common/src/main/java/org/apache/hbase/io/codec/OldSchoolKeyValueEncoder.java b/hbase-common/src/main/java/org/apache/hbase/io/codec/OldSchoolKeyValueEncoder.java
new file mode 100644
index 0000000..dffac45
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hbase/io/codec/OldSchoolKeyValueEncoder.java
@@ -0,0 +1,68 @@
+package org.apache.hbase.io.codec;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hbase.Cell;
+
+/**
+ * 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. When {@link #finish} is called,
+ * we write out an End-Of-KeyValues marker to the stream. If you wrote two KeyValues to
+ * this encoder, it would look like this in the stream:
+ *
+ * length-of-KeyValue1 // A java int with the length of KeyValue1 backing array + * KeyValue1 backing array + * length-of-KeyValue2 + * KeyValue2 backing array + * length-of-zero // A java int whose value is 0; marks end of the encoded section + *+ * @see OldSchoolKeyValueDecoder + */ +public class OldSchoolKeyValueEncoder implements Encoder { + // We write out an '0' int as marker that there are no more kvs when you call flush. + static final int END_OF_KEYVALUES = 0; + // Need to be able to write java types such as int and long so need DataOutput. + // Want to stream too so DataOutputStream. + private final DataOutputStream out; + // This encoder is 'done' once flush has been called because on flush we + // write out the END_OF_KEYVALUES marker. + private boolean finish = false; + + public OldSchoolKeyValueEncoder(final DataOutputStream out) { + this.out = out; + } + + @Override + public Cell encode(Cell cell) throws CodecException { + if (this.finish) throw new CodecException("Finished"); + // This is crass and will not work when KV changes + try { + KeyValue.write((KeyValue)cell, this.out); + } catch (IOException e) { + throw new CodecException(e); + } + return cell; + } + + /** + * Calling flush 'finishes' this encoder. Subsequent calls + * to {@link #write(Cell)} will throw exception. + */ + @Override + public void finish() throws CodecException { + if (this.finish) return; + this.finish = true; + // Write out an int whose value is zero as end of stream. + try { + this.out.writeInt(END_OF_KEYVALUES); + this.out.flush(); + } catch (IOException e) { + throw new CodecException(e); + } + } +} \ No newline at end of file diff --git a/hbase-common/src/test/java/org/apache/hbase/io/codec/TestCellCodec.java b/hbase-common/src/test/java/org/apache/hbase/io/codec/TestCellCodec.java new file mode 100644 index 0000000..2cd6355 --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hbase/io/codec/TestCellCodec.java @@ -0,0 +1,88 @@ +package org.apache.hbase.io.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.util.Bytes; +import org.apache.hbase.Cell; +import org.apache.hbase.CellComparator; +import org.junit.Test; + +import com.google.common.io.CountingInputStream; +import com.google.common.io.CountingOutputStream; + +public class TestCellCodec { + + @Test + public void testEmptyWorks() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + CellEncoder encoder = new CellEncoder(dos); + encoder.finish(); + dos.close(); + long offset = cos.getCount(); + assertEquals(0 + 1 /*CellEncoder.END_OF_CELLS*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + CellDecoder decoder = new CellDecoder(dis); + assertTrue(decoder.decode() == null); + dis.close(); + assertEquals(0 + 1 /*CellEncoder.END_OF_CELLS*/, cis.getCount()); + } + + @Test + public void testOne() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + CellEncoder encoder = new CellEncoder(dos); + final KeyValue kv = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("v")); + encoder.encode(kv); + encoder.finish(); + dos.close(); + long offset = cos.getCount(); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + CellDecoder decoder = new CellDecoder(dis); + assertTrue(decoder.decode() != null); // First read should pull in the KV + assertTrue(decoder.decode() == null); // Second read should trip over the end-of-stream marker and return false + 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); + CellEncoder encoder = new CellEncoder(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.encode(kv1); + encoder.encode(kv2); + encoder.encode(kv3); + encoder.finish(); + dos.close(); + long offset = cos.getCount(); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + CellDecoder decoder = new CellDecoder(dis); + Cell c = decoder.decode(); + assertTrue(CellComparator.equals(c, kv1)); + c = decoder.decode(); + assertTrue(CellComparator.equals(c, kv2)); + c = decoder.decode(); + assertTrue(CellComparator.equals(c, kv3)); + assertTrue(decoder.decode() == null); + dis.close(); + assertEquals(offset, cis.getCount()); + } +} \ No newline at end of file diff --git a/hbase-common/src/test/java/org/apache/hbase/io/codec/TestOldSchoolKeyValueCodec.java b/hbase-common/src/test/java/org/apache/hbase/io/codec/TestOldSchoolKeyValueCodec.java new file mode 100644 index 0000000..6af2542 --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hbase/io/codec/TestOldSchoolKeyValueCodec.java @@ -0,0 +1,93 @@ +package org.apache.hbase.io.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.util.Bytes; +import org.junit.Test; + +import com.google.common.io.CountingInputStream; +import com.google.common.io.CountingOutputStream; + +public class TestOldSchoolKeyValueCodec { + + @Test + public void testEmptyWorks() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + OldSchoolKeyValueEncoder encoder = new OldSchoolKeyValueEncoder(dos); + encoder.finish(); + dos.close(); + long offset = cos.getCount(); + assertEquals(0 + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + OldSchoolKeyValueDecoder decoder = new OldSchoolKeyValueDecoder(dis); + assertTrue(decoder.decode() == null); + dis.close(); + assertEquals(0 + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } + + @Test + public void testOne() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + OldSchoolKeyValueEncoder encoder = new OldSchoolKeyValueEncoder(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.encode(kv); + encoder.finish(); + dos.close(); + long offset = cos.getCount(); + assertEquals(length + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + OldSchoolKeyValueDecoder decoder = new OldSchoolKeyValueDecoder(dis); + assertTrue(decoder.decode() != null); // First read should pull in the KV + assertTrue(decoder.decode() == null); // Second read should trip over the end-of-stream marker and return false + dis.close(); + assertEquals(length + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } + + @Test + public void testThree() throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CountingOutputStream cos = new CountingOutputStream(baos); + DataOutputStream dos = new DataOutputStream(cos); + OldSchoolKeyValueEncoder encoder = new OldSchoolKeyValueEncoder(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.encode(kv1); + encoder.encode(kv2); + encoder.encode(kv3); + encoder.finish(); + dos.close(); + long offset = cos.getCount(); + assertEquals(length * 3 + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, offset); + CountingInputStream cis = new CountingInputStream(new ByteArrayInputStream(baos.toByteArray())); + DataInputStream dis = new DataInputStream(cis); + OldSchoolKeyValueDecoder decoder = new OldSchoolKeyValueDecoder(dis); + KeyValue kv = (KeyValue)decoder.decode(); + assertTrue(kv1.equals(kv)); + kv = (KeyValue)decoder.decode(); + assertTrue(kv2.equals(kv)); + kv = (KeyValue)decoder.decode(); + assertTrue(kv3.equals(kv)); + assertTrue(decoder.decode() == null); + dis.close(); + assertEquals((length * 3) + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } +} \ No newline at end of file