diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java index 85ebc71..9e576fa 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/KeyValue.java @@ -37,7 +37,7 @@ import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.ClassSize; 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; @@ -918,7 +918,7 @@ public class KeyValue implements Cell, HeapSize { return "empty"; } return keyToString(this.bytes, this.offset + ROW_OFFSET, getKeyLength()) + - "/vlen=" + getValueLength() + "/ts=" + memstoreTS; + "/vlen=" + getValueLength() + "/mvcc=" + memstoreTS; } /** @@ -2289,8 +2289,10 @@ public class KeyValue implements Cell, HeapSize { } /** - * @param in Where to read bytes from - * @return KeyValue created by deserializing from 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 Comparator, 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){ + if (!areKeyLengthsEqual(a, b)) { + return false; + } + //TODO compare byte[]'s in reverse since later bytes more likely to differ + return 0 == compareStatic(a, b); + } + + public static boolean equalsRow(Cell a, Cell b){ + if(!areRowLengthsEqual(a, b)){ + return false; + } + return 0 == Bytes.compareTo( + a.getRowArray(), a.getRowOffset(), a.getRowLength(), + b.getRowArray(), b.getRowOffset(), b.getRowLength()); + } + + + /********************* 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); + } + +} diff --git a/hbase-common/src/main/java/org/apache/hbase/CellScanner.java b/hbase-common/src/main/java/org/apache/hbase/CellScanner.java new file mode 100644 index 0000000..f6d1128 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/CellScanner.java @@ -0,0 +1,66 @@ +/** + * 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. + * 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 {@link #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 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. Use + * {@link #getDeepCopy()} if you want a Cell that does not hold references (if the data is + * encoded or compressed, the call to {@link #getDeepCopy()} will cost more than a {@link #get()}. + *

+ * Typical usage: + * + *

+ * while (scanner.next()) {
+ *   Cell cell = scanner.get();
+ *   // do something
+ * }
+ * 
+ */ +@InterfaceAudience.Private +@InterfaceStability.Unstable +public interface CellScanner { + /** + * @return the current Cell which may be mutable + */ + Cell get(); + + /** + * @return Get a deep copy of the current Cell or null if no current Cell + */ + Cell getDeepCopy(); + + /** + * Advance the scanner 1 cell. + * @return true if the next cell is found and {@link #get()} will return a valid Cell + */ + boolean next(); +} \ No newline at end of file diff --git a/hbase-common/src/main/java/org/apache/hbase/CellTool.java b/hbase-common/src/main/java/org/apache/hbase/CellTool.java new file mode 100644 index 0000000..96bb509 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/CellTool.java @@ -0,0 +1,117 @@ +/* + * 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 org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.hbase.util.ByteRange; + +@InterfaceAudience.Private +@InterfaceStability.Evolving +public final class CellTool { + + /******************* 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]; + } + + + /********************** KeyValue (move to KeyValueUtils) *********************/ + + 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; + } + +} diff --git a/hbase-common/src/main/java/org/apache/hbase/cell/CellComparator.java b/hbase-common/src/main/java/org/apache/hbase/cell/CellComparator.java deleted file mode 100644 index 19d6b54..0000000 --- a/hbase-common/src/main/java/org/apache/hbase/cell/CellComparator.java +++ /dev/null @@ -1,178 +0,0 @@ -/* - * 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.cell; - -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 org.apache.hbase.Cell; - -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 Comparator, 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){ - if (!areKeyLengthsEqual(a, b)) { - return false; - } - //TODO compare byte[]'s in reverse since later bytes more likely to differ - return 0 == compareStatic(a, b); - } - - public static boolean equalsRow(Cell a, Cell b){ - if(!areRowLengthsEqual(a, b)){ - return false; - } - return 0 == Bytes.compareTo( - a.getRowArray(), a.getRowOffset(), a.getRowLength(), - b.getRowArray(), b.getRowOffset(), b.getRowLength()); - } - - - /********************* 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); - } - -} diff --git a/hbase-common/src/main/java/org/apache/hbase/cell/CellOutputStream.java b/hbase-common/src/main/java/org/apache/hbase/cell/CellOutputStream.java deleted file mode 100644 index fcaf27e..0000000 --- a/hbase-common/src/main/java/org/apache/hbase/cell/CellOutputStream.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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.cell; - -import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hbase.Cell; - -/** - * Accepts a stream of Cells and adds them to its internal data structure. 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, but more efficient implementations will append results to a - * byte[] to eliminate overhead, and possibly encode the cells further. - */ -@InterfaceAudience.Private -@InterfaceStability.Evolving -public interface CellOutputStream { - - /** - * Implementation must copy the entire state of the Cell. If the appended Cell is modified - * immediately after the append method returns, the modifications must have absolutely no effect - * on the copy of the Cell that was added to the appender. For example, calling someList.add(cell) - * is not correct. - */ - void write(Cell cell); - - /** - * 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. - */ - void flush(); - -} diff --git a/hbase-common/src/main/java/org/apache/hbase/cell/CellScannerPosition.java b/hbase-common/src/main/java/org/apache/hbase/cell/CellScannerPosition.java deleted file mode 100644 index eeadf5f..0000000 --- a/hbase-common/src/main/java/org/apache/hbase/cell/CellScannerPosition.java +++ /dev/null @@ -1,68 +0,0 @@ -/** - * 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.cell; - -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: - *
    - *
  • In a DataBlockScanner, the AFTER_LAST position indicates to the parent StoreFileScanner that - * it should load the next block.
  • - *
  • In a StoreFileScanner, the AFTER_LAST position indicates that the file has been exhausted.
  • - *
  • In a RegionScanner, the AFTER_LAST position indicates that the scanner should move to the - * next region.
  • - *
- */ -@InterfaceAudience.Public -@InterfaceStability.Evolving -public enum CellScannerPosition { - - /** - * getCurrentCell() will NOT return a valid cell. Calling nextCell() will advance to the first - * cell. - */ - BEFORE_FIRST, - - /** - * getCurrentCell() will return a valid cell, but it is not the cell requested by positionAt(..), - * rather it is the nearest cell before the requested cell. - */ - BEFORE, - - /** - * getCurrentCell() will return a valid cell, and it is exactly the cell that was requested by - * positionAt(..). - */ - AT, - - /** - * getCurrentCell() will return a valid cell, but it is not the cell requested by positionAt(..), - * rather it is the nearest cell after the requested cell. - */ - AFTER, - - /** - * getCurrentCell() will NOT return a valid cell. Calling nextCell() will have no effect. - */ - AFTER_LAST - -} diff --git a/hbase-common/src/main/java/org/apache/hbase/cell/CellTool.java b/hbase-common/src/main/java/org/apache/hbase/cell/CellTool.java deleted file mode 100644 index 229ca36..0000000 --- a/hbase-common/src/main/java/org/apache/hbase/cell/CellTool.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * 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.cell; - -import java.nio.ByteBuffer; - -import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.hbase.util.ByteRange; -import org.apache.hbase.Cell; - -@InterfaceAudience.Private -@InterfaceStability.Evolving -public final class CellTool { - - /******************* 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]; - } - - - /********************** KeyValue (move to KeyValueUtils) *********************/ - - 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; - } - -} diff --git a/hbase-common/src/main/java/org/apache/hbase/codec/CellDecoder.java b/hbase-common/src/main/java/org/apache/hbase/codec/CellDecoder.java new file mode 100644 index 0000000..53a0afb --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/codec/CellDecoder.java @@ -0,0 +1,77 @@ +package org.apache.hbase.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; +import org.apache.hbase.CellScanner; + +/** + * @see CellEncoder + */ +public class CellDecoder implements CellScanner { + private final DataInputStream in; + private boolean hasNext = true; + private Cell current = null; + + public CellDecoder(final DataInputStream in) { + this.in = in; + } + + @Override + public boolean next() { + if (!this.hasNext) return this.hasNext; + int length; + try { + length = WritableUtils.readVInt(this.in); + } catch (IOException e) { + throw new RuntimeException(e); + } + if (length == CellEncoder.END_OF_CELLS) { + this.hasNext = false; + } 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. + // TODO: Make a new Cell implementation that just carries these + // byte arrays. + this.current = new KeyValue(row, family, qualifier, timestamp, + KeyValue.Type.codeToType(type), value); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + return this.hasNext; + } + + /** + * @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; + } + + @Override + public Cell get() { + return this.current; + } + + @Override + public Cell getDeepCopy() { + return get(); + } +} \ No newline at end of file diff --git a/hbase-common/src/main/java/org/apache/hbase/codec/CellEncoder.java b/hbase-common/src/main/java/org/apache/hbase/codec/CellEncoder.java new file mode 100644 index 0000000..373bee0 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/codec/CellEncoder.java @@ -0,0 +1,82 @@ +package org.apache.hbase.codec; + +import java.io.DataOutputStream; +import java.io.IOException; + +import org.apache.hadoop.io.WritableUtils; +import org.apache.hbase.Cell; +import org.apache.hbase.io.CellOutputStream; + +/** + * 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 CellOutputStream { + // TODO: Replace WritableUtils from hadoop vint w/ our own or something faster? + // TOOD: I need a CellFactory to make Cells with. Using KeyValue for now. + // TOOD: Can I have a Cell implementation that is backed by these little + // byte arrays I made? Or one that reads off the steam and puts bytes into + // a ByteBuffer? + + // 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 void write(Cell cell) throws IOException { + 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); + } + } + + /** + * 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 flush() throws IOException { + 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/codec/CodecException.java b/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java new file mode 100644 index 0000000..d3f5bff --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/codec/CodecException.java @@ -0,0 +1,22 @@ +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 diff --git a/hbase-common/src/main/java/org/apache/hbase/codec/OldSchoolKeyValueDecoder.java b/hbase-common/src/main/java/org/apache/hbase/codec/OldSchoolKeyValueDecoder.java new file mode 100644 index 0000000..1a66550 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/codec/OldSchoolKeyValueDecoder.java @@ -0,0 +1,52 @@ +package org.apache.hbase.codec; + +import java.io.DataInputStream; +import java.io.IOException; + +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hbase.Cell; +import org.apache.hbase.CellScanner; + +/** + * @see OldSchoolKeyValueEncoder + */ +public class OldSchoolKeyValueDecoder implements CellScanner { + private final DataInputStream in; + private boolean hasNext = true; + private Cell current = null; + + public OldSchoolKeyValueDecoder(final DataInputStream in) { + this.in = in; + } + + @Override + public boolean next() { + if (!this.hasNext) return !this.hasNext; + int length; + try { + length = this.in.readInt(); + } catch (IOException e) { + throw new RuntimeException(e); + } + if (length == OldSchoolKeyValueEncoder.END_OF_KEYVALUES) { + this.hasNext = false; + } else { + try { + this.current = KeyValue.create(length, this.in); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + return this.hasNext; + } + + @Override + public Cell get() { + return this.current; + } + + @Override + public Cell getDeepCopy() { + return get(); + } +} \ No newline at end of file diff --git a/hbase-common/src/main/java/org/apache/hbase/codec/OldSchoolKeyValueEncoder.java b/hbase-common/src/main/java/org/apache/hbase/codec/OldSchoolKeyValueEncoder.java new file mode 100644 index 0000000..ebb2483 --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/codec/OldSchoolKeyValueEncoder.java @@ -0,0 +1,70 @@ +package org.apache.hbase.codec; + +import java.io.DataOutputStream; +import java.io.IOException; + +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hbase.Cell; +import org.apache.hbase.io.CellOutputStream; + +/** + * 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 #flush} is called, + * we write out an End-Of-KeyValues marker to the stream and subsequent {@link #write(Cell)} + * calls will throw an exception. + * + *

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 filled with a KeyValue serialized in its particular format
+ * 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 CellOutputStream { + // 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 flush = false; + + public OldSchoolKeyValueEncoder(final DataOutputStream out) { + this.out = out; + } + + @Override + public void write(Cell cell) throws IOException { + if (this.flush) 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); + } + } + + /** + * Calling flush 'finishes' this encoder. Subsequent calls + * to {@link #write(Cell)} will throw exception. + */ + @Override + public void flush() throws IOException { + if (this.flush) return; + this.flush = 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/main/java/org/apache/hbase/io/CellOutputStream.java b/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java new file mode 100644 index 0000000..b42005f --- /dev/null +++ b/hbase-common/src/main/java/org/apache/hbase/io/CellOutputStream.java @@ -0,0 +1,54 @@ +/* + * 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; + +/** + * 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, but more efficient implementations will append results to a + * byte[] to eliminate overhead, and possibly encode the cells further. + *

To read Cells, use {@link CellInputStream} + * @see CellInputStream + */ +@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/test/java/org/apache/hbase/codec/TestCellCodec.java b/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java new file mode 100644 index 0000000..458bd8a --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hbase/codec/TestCellCodec.java @@ -0,0 +1,93 @@ +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.util.Bytes; +import org.apache.hbase.Cell; +import org.apache.hbase.CellComparator; +import org.apache.hbase.codec.CellDecoder; +import org.apache.hbase.codec.CellEncoder; +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.flush(); + 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); + assertFalse(decoder.next()); + 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.write(kv); + encoder.flush(); + 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.next()); // First read should pull in the KV + assertFalse(decoder.next()); // 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.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); + CellDecoder decoder = new CellDecoder(dis); + assertTrue(decoder.next()); + Cell c = decoder.get(); + assertTrue(CellComparator.equals(c, kv1)); + assertTrue(decoder.next()); + c = decoder.get(); + assertTrue(CellComparator.equals(c, kv2)); + assertTrue(decoder.next()); + c = decoder.get(); + assertTrue(CellComparator.equals(c, kv3)); + assertFalse(decoder.next()); + dis.close(); + assertEquals(offset, cis.getCount()); + } +} \ No newline at end of file diff --git a/hbase-common/src/test/java/org/apache/hbase/codec/TestOldSchoolKeyValueCodec.java b/hbase-common/src/test/java/org/apache/hbase/codec/TestOldSchoolKeyValueCodec.java new file mode 100644 index 0000000..af788c1 --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hbase/codec/TestOldSchoolKeyValueCodec.java @@ -0,0 +1,98 @@ +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.util.Bytes; +import org.apache.hbase.codec.OldSchoolKeyValueDecoder; +import org.apache.hbase.codec.OldSchoolKeyValueEncoder; +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.flush(); + 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); + assertFalse(decoder.next()); + 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.write(kv); + encoder.flush(); + 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.next()); // First read should pull in the KV + assertFalse(decoder.next()); // 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.write(kv1); + encoder.write(kv2); + encoder.write(kv3); + encoder.flush(); + 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); + assertTrue(decoder.next()); + KeyValue kv = (KeyValue)decoder.get(); + assertTrue(kv1.equals(kv)); + assertTrue(decoder.next()); + kv = (KeyValue)decoder.get(); + assertTrue(kv2.equals(kv)); + assertTrue(decoder.next()); + kv = (KeyValue)decoder.get(); + assertTrue(kv3.equals(kv)); + assertFalse(decoder.next()); + dis.close(); + assertEquals((length * 3) + Bytes.SIZEOF_INT /*OldSchoolKeyValueEncoder.END_OF_KEYVALUES*/, + cis.getCount()); + } +} \ No newline at end of file