hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anoopsamj...@apache.org
Subject hbase git commit: HBASE-16213 A new HFileBlock structure for fast random get. (binlijin)
Date Mon, 29 Aug 2016 06:42:10 GMT
Repository: hbase
Updated Branches:
  refs/heads/branch-1 35fa34191 -> c899897bc


HBASE-16213 A new HFileBlock structure for fast random get. (binlijin)


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/c899897b
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/c899897b
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/c899897b

Branch: refs/heads/branch-1
Commit: c899897bc8dc4a7eccc9e2a80fd05ad55654f18e
Parents: 35fa341
Author: anoopsamjohn <anoopsamjohn@gmail.com>
Authored: Mon Aug 29 12:11:46 2016 +0530
Committer: anoopsamjohn <anoopsamjohn@gmail.com>
Committed: Mon Aug 29 12:11:46 2016 +0530

----------------------------------------------------------------------
 .../hadoop/hbase/io/ByteArrayOutputStream.java  | 127 ++++++
 .../hbase/io/encoding/DataBlockEncoding.java    |   3 +-
 .../hbase/io/encoding/RowIndexCodecV1.java      | 165 +++++++
 .../hbase/io/encoding/RowIndexEncoderV1.java    | 115 +++++
 .../hbase/io/encoding/RowIndexSeekerV1.java     | 431 +++++++++++++++++++
 .../encoding/TestSeekToBlockWithEncoders.java   |   4 +-
 6 files changed, 842 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/c899897b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ByteArrayOutputStream.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ByteArrayOutputStream.java
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ByteArrayOutputStream.java
new file mode 100644
index 0000000..a3c571f
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/ByteArrayOutputStream.java
@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.io;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.BufferOverflowException;
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Our own implementation of ByteArrayOutputStream where all methods are NOT
+ * synchronized and supports writing ByteBuffer directly to it.
+ */
+@InterfaceAudience.Private
+public class ByteArrayOutputStream extends OutputStream {
+
+  // Borrowed from openJDK:
+  // http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/ArrayList.java#221
+  private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
+
+  private byte[] buf;
+  private int pos = 0;
+
+  public ByteArrayOutputStream() {
+    this(32);
+  }
+
+  public ByteArrayOutputStream(int capacity) {
+    this.buf = new byte[capacity];
+  }
+
+  /**
+   * Writes an <code>int</code> to the underlying output stream as four
+   * bytes, high byte first.
+   * @param i the <code>int</code> to write
+   * @throws IOException if an I/O error occurs.
+  */
+  public void writeInt(int i) throws IOException {
+    checkSizeAndGrow(Bytes.SIZEOF_INT);
+    Bytes.putInt(this.buf, this.pos, i);
+    this.pos += Bytes.SIZEOF_INT;
+  }
+
+  @Override
+  public void write(int b) throws IOException {
+    checkSizeAndGrow(Bytes.SIZEOF_BYTE);
+    buf[this.pos] = (byte) b;
+    this.pos++;
+  }
+
+  @Override
+  public void write(byte[] b, int off, int len) throws IOException {
+    checkSizeAndGrow(len);
+    System.arraycopy(b, off, this.buf, this.pos, len);
+    this.pos += len;
+  }
+
+  private void checkSizeAndGrow(int extra) {
+    long capacityNeeded = this.pos + (long) extra;
+    if (capacityNeeded > this.buf.length) {
+      // guarantee it's possible to fit
+      if (capacityNeeded > MAX_ARRAY_SIZE) {
+        throw new BufferOverflowException();
+      }
+      // double until hit the cap
+      long nextCapacity = Math.min(this.buf.length << 1, MAX_ARRAY_SIZE);
+      // but make sure there is enough if twice the existing capacity is still
+      // too small
+      nextCapacity = Math.max(nextCapacity, capacityNeeded);
+      if (nextCapacity > MAX_ARRAY_SIZE) {
+        throw new BufferOverflowException();
+      }
+      byte[] newBuf = new byte[(int) nextCapacity];
+      System.arraycopy(buf, 0, newBuf, 0, buf.length);
+      buf = newBuf;
+    }
+  }
+
+  /**
+   * Resets the <code>pos</code> field of this byte array output stream to zero.
+   * The output stream can be used again.
+   */
+  public void reset() {
+    this.pos = 0;
+  }
+
+  /**
+   * Copies the content of this Stream into a new byte array.
+   *
+   * @return the contents of this output stream, as new byte array.
+   */
+  public byte toByteArray()[] {
+    return Arrays.copyOf(buf, pos);
+  }
+
+  /**
+   * @return the underlying array where the data gets accumulated
+   */
+  public byte[] getBuffer() {
+    return this.buf;
+  }
+
+  /**
+   * @return The current size of the buffer.
+   */
+  public int size() {
+    return this.pos;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/c899897b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DataBlockEncoding.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DataBlockEncoding.java
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DataBlockEncoding.java
index 67d18ed..71b55e2 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DataBlockEncoding.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DataBlockEncoding.java
@@ -43,7 +43,8 @@ public enum DataBlockEncoding {
   FAST_DIFF(4, "org.apache.hadoop.hbase.io.encoding.FastDiffDeltaEncoder"),
   // id 5 is reserved for the COPY_KEY algorithm for benchmarking
   // COPY_KEY(5, "org.apache.hadoop.hbase.io.encoding.CopyKeyDataBlockEncoder"),
-  PREFIX_TREE(6, "org.apache.hadoop.hbase.codec.prefixtree.PrefixTreeCodec");
+  PREFIX_TREE(6, "org.apache.hadoop.hbase.codec.prefixtree.PrefixTreeCodec"),
+  ROW_INDEX_V1(7, "org.apache.hadoop.hbase.io.encoding.RowIndexCodecV1");
 
   private final short id;
   private final byte[] idInBytes;

http://git-wip-us.apache.org/repos/asf/hbase/blob/c899897b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java
new file mode 100644
index 0000000..f18e094
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hadoop.hbase.io.encoding;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.KVComparator;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.io.hfile.BlockType;
+import org.apache.hadoop.hbase.io.hfile.HFileContext;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Store cells following every row's start offset, so we can binary search to a row's cells.
+ *
+ * Format:
+ * flat cells
+ * integer: number of rows
+ * integer: row0's offset
+ * integer: row1's offset
+ * ....
+ * integer: dataSize
+ *
+*/
+@InterfaceAudience.Private
+public class RowIndexCodecV1 implements DataBlockEncoder {
+
+  private static class RowIndexEncodingState extends EncodingState {
+    RowIndexEncoderV1 encoder = null;
+  }
+
+  @Override
+  public void startBlockEncoding(HFileBlockEncodingContext blkEncodingCtx,
+      DataOutputStream out) throws IOException {
+    if (blkEncodingCtx.getClass() != HFileBlockDefaultEncodingContext.class) {
+      throw new IOException(this.getClass().getName() + " only accepts "
+          + HFileBlockDefaultEncodingContext.class.getName() + " as the "
+          + "encoding context.");
+    }
+
+    HFileBlockDefaultEncodingContext encodingCtx = (HFileBlockDefaultEncodingContext) blkEncodingCtx;
+    encodingCtx.prepareEncoding(out);
+
+    RowIndexEncoderV1 encoder = new RowIndexEncoderV1(out, encodingCtx);
+    RowIndexEncodingState state = new RowIndexEncodingState();
+    state.encoder = encoder;
+    blkEncodingCtx.setEncodingState(state);
+  }
+
+  @Override
+  public int encode(Cell cell, HFileBlockEncodingContext encodingCtx,
+      DataOutputStream out) throws IOException {
+    RowIndexEncodingState state = (RowIndexEncodingState) encodingCtx
+        .getEncodingState();
+    RowIndexEncoderV1 encoder = state.encoder;
+    return encoder.write(cell);
+  }
+
+  @Override
+  public void endBlockEncoding(HFileBlockEncodingContext encodingCtx,
+      DataOutputStream out, byte[] uncompressedBytesWithHeader)
+      throws IOException {
+    RowIndexEncodingState state = (RowIndexEncodingState) encodingCtx
+        .getEncodingState();
+    RowIndexEncoderV1 encoder = state.encoder;
+    encoder.flush();
+    if (encodingCtx.getDataBlockEncoding() != DataBlockEncoding.NONE) {
+      encodingCtx.postEncoding(BlockType.ENCODED_DATA);
+    } else {
+      encodingCtx.postEncoding(BlockType.DATA);
+    }
+  }
+
+  @Override
+  public ByteBuffer decodeKeyValues(DataInputStream source,
+      HFileBlockDecodingContext decodingCtx) throws IOException {
+    if (!decodingCtx.getHFileContext().isIncludesTags()) {
+      ByteBuffer sourceAsBuffer = ByteBufferUtils
+          .drainInputStreamToBuffer(source);// waste
+      sourceAsBuffer.mark();
+      sourceAsBuffer.position(sourceAsBuffer.limit() - Bytes.SIZEOF_INT);
+      int onDiskSize = sourceAsBuffer.getInt();
+      sourceAsBuffer.reset();
+      ByteBuffer dup = sourceAsBuffer.duplicate();
+      dup.position(sourceAsBuffer.position());
+      dup.limit(sourceAsBuffer.position() + onDiskSize);
+      return dup.slice();
+    } else {
+      ByteBuffer sourceAsBuffer = ByteBufferUtils
+          .drainInputStreamToBuffer(source);// waste
+      sourceAsBuffer.mark();
+      RowIndexSeekerV1 seeker = new RowIndexSeekerV1(KeyValue.COMPARATOR,
+          decodingCtx);
+      seeker.setCurrentBuffer(sourceAsBuffer);
+      List<ByteBuffer> kvs = new ArrayList<ByteBuffer>();
+      kvs.add(seeker.getKeyValueBuffer());
+      while (seeker.next()) {
+        kvs.add(seeker.getKeyValueBuffer());
+      }
+      int totalLength = 0;
+      for (ByteBuffer buf : kvs) {
+        totalLength += buf.remaining();
+      }
+      byte[] keyValueBytes = new byte[totalLength];
+      ByteBuffer result = ByteBuffer.wrap(keyValueBytes);
+      for (ByteBuffer buf : kvs) {
+        result.put(buf);
+      }
+      return result;
+    }
+  }
+
+  @Override
+  public ByteBuffer getFirstKeyInBlock(ByteBuffer block) {
+    block.mark();
+    int keyLength = block.getInt();
+    block.getInt();
+    int pos = block.position();
+    block.reset();
+    ByteBuffer dup = block.duplicate();
+    dup.position(pos);
+    dup.limit(pos + keyLength);
+    return dup.slice();
+  }
+
+  @Override
+  public EncodedSeeker createSeeker(KVComparator comparator,
+      HFileBlockDecodingContext decodingCtx) {
+    return new RowIndexSeekerV1(comparator, decodingCtx);
+  }
+
+  @Override
+  public HFileBlockEncodingContext newDataBlockEncodingContext(
+      DataBlockEncoding encoding, byte[] header, HFileContext meta) {
+    return new HFileBlockDefaultEncodingContext(encoding, header, meta);
+  }
+
+  @Override
+  public HFileBlockDecodingContext newDataBlockDecodingContext(HFileContext meta) {
+    return new HFileBlockDefaultDecodingContext(meta);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/c899897b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexEncoderV1.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexEncoderV1.java
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexEncoderV1.java
new file mode 100644
index 0000000..888ef9e
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexEncoderV1.java
@@ -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.hadoop.hbase.io.encoding;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueUtil;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
+import org.apache.hadoop.io.WritableUtils;
+
+@InterfaceAudience.Private
+public class RowIndexEncoderV1 {
+  private static final Log LOG = LogFactory.getLog(RowIndexEncoderV1.class);
+
+  /** The Cell previously appended. */
+  private Cell lastCell = null;
+
+  private DataOutputStream out;
+  private HFileBlockDefaultEncodingContext encodingCtx;
+  private int startOffset = -1;
+  private ByteArrayOutputStream rowsOffsetBAOS = new ByteArrayOutputStream(
+      64 * 4);
+
+  public RowIndexEncoderV1(DataOutputStream out,
+      HFileBlockDefaultEncodingContext encodingCtx) {
+    this.out = out;
+    this.encodingCtx = encodingCtx;
+  }
+
+  public int write(Cell cell) throws IOException {
+    // checkKey uses comparator to check we are writing in order.
+    if (!checkRow(cell)) {
+      if (startOffset < 0) {
+        startOffset = out.size();
+      }
+      rowsOffsetBAOS.writeInt(out.size() - startOffset);
+    }
+    int klength = KeyValueUtil.keyLength(cell);
+    int vlength = cell.getValueLength();
+    out.writeInt(klength);
+    out.writeInt(vlength);
+    CellUtil.writeFlatKey(cell, out);
+    // Write the value part
+    out.write(cell.getValueArray(), cell.getValueOffset(), vlength);
+    int encodedKvSize = klength + vlength
+        + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE;
+    // Write the additional tag into the stream
+    if (encodingCtx.getHFileContext().isIncludesTags()) {
+      int tagsLength = cell.getTagsLength();
+      out.writeShort(tagsLength);
+      // There are some tags to be written
+      if (tagsLength > 0) {
+        out.write(cell.getTagsArray(), cell.getTagsOffset(), tagsLength);
+      }
+      encodedKvSize += tagsLength + KeyValue.TAGS_LENGTH_SIZE;
+    }
+    if (encodingCtx.getHFileContext().isIncludesMvcc()) {
+      WritableUtils.writeVLong(out, cell.getSequenceId());
+      encodedKvSize += WritableUtils.getVIntSize(cell.getSequenceId());
+    }
+    lastCell = cell;
+    return encodedKvSize;
+  }
+
+  protected boolean checkRow(final Cell cell) throws IOException {
+    boolean isDuplicateRow = false;
+    if (cell == null) {
+      throw new IOException("Key cannot be null or empty");
+    }
+    if (lastCell != null) {
+      int keyComp = KeyValue.COMPARATOR.compareRows(lastCell, cell);
+      if (keyComp > 0) {
+        throw new IOException("Added a key not lexically larger than"
+            + " previous. Current cell = " + cell + ", lastCell = " + lastCell);
+      } else if (keyComp == 0) {
+        isDuplicateRow = true;
+      }
+    }
+    return isDuplicateRow;
+  }
+
+  public void flush() throws IOException {
+    int onDiskDataSize = 0;
+    if (startOffset >= 0) {
+      onDiskDataSize = out.size() - startOffset;
+    }
+    // rowsOffsetBAOS.size() / 4
+    out.writeInt(rowsOffsetBAOS.size() >> 2);
+    if (rowsOffsetBAOS.size() > 0) {
+      out.write(rowsOffsetBAOS.getBuffer(), 0, rowsOffsetBAOS.size());
+    }
+    out.writeInt(onDiskDataSize);
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("RowNumber: " + (rowsOffsetBAOS.size() >> 2)
+          + ", onDiskDataSize: " + onDiskDataSize + ", totalOnDiskSize: "
+          + (out.size() - startOffset));
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/c899897b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexSeekerV1.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexSeekerV1.java
b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexSeekerV1.java
new file mode 100644
index 0000000..a3289d6
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexSeekerV1.java
@@ -0,0 +1,431 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hadoop.hbase.io.encoding;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.KVComparator;
+import org.apache.hadoop.hbase.NoTagsKeyValue;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoder.EncodedSeeker;
+import org.apache.hadoop.hbase.util.ByteBufferUtils;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.SimpleMutableByteRange;
+import org.apache.hadoop.io.WritableUtils;
+
+@InterfaceAudience.Private
+public class RowIndexSeekerV1 implements EncodedSeeker {
+
+  private HFileBlockDecodingContext decodingCtx;
+  private final KVComparator comparator;
+
+  private ByteBuffer currentBuffer;
+  private SeekerState current = new SeekerState(); // always valid
+  private SeekerState previous = new SeekerState(); // may not be valid
+
+  private int rowNumber;
+  private ByteBuffer rowOffsets = null;
+
+  public RowIndexSeekerV1(KVComparator comparator,
+      HFileBlockDecodingContext decodingCtx) {
+    this.comparator = comparator;
+    this.decodingCtx = decodingCtx;
+  }
+
+  @Override
+  public void setCurrentBuffer(ByteBuffer buffer) {
+    int onDiskSize = Bytes.toIntUnsafe(buffer.array(), buffer.arrayOffset()
+        + buffer.limit() - Bytes.SIZEOF_INT);
+    // int onDiskSize = buffer.getInt(buffer.limit() - Bytes.SIZEOF_INT);
+
+    // Data part
+    ByteBuffer dup = buffer.duplicate();
+    dup.position(buffer.position());
+    dup.limit(buffer.position() + onDiskSize);
+    currentBuffer = dup.slice();
+    current.currentBuffer = currentBuffer;
+    ByteBufferUtils.skip(buffer, onDiskSize);
+
+    // Row offset
+    rowNumber = buffer.getInt();
+    // equals Bytes.SIZEOF_INT * rowNumber
+    int totalRowOffsetsLength = rowNumber << 2;
+    ByteBuffer rowDup = buffer.duplicate();
+    rowDup.position(buffer.position());
+    rowDup.limit(buffer.position() + totalRowOffsetsLength);
+    rowOffsets = rowDup.slice();
+
+    decodeFirst();
+  }
+
+  @Override
+  public ByteBuffer getKeyDeepCopy() {
+    ByteBuffer keyBuffer = ByteBuffer.allocate(current.keyLength);
+    keyBuffer.put(current.keyBuffer.getBytes(), current.keyBuffer.getOffset(),
+        current.keyLength);
+    keyBuffer.rewind();
+    return keyBuffer;
+  }
+
+  @Override
+  public ByteBuffer getValueShallowCopy() {
+    ByteBuffer dup = currentBuffer.duplicate();
+    dup.position(current.valueOffset);
+    dup.limit(current.valueOffset + current.valueLength);
+    return dup.slice();
+  }
+
+  ByteBuffer getKeyValueBuffer() {
+    ByteBuffer kvBuffer = createKVBuffer();
+    kvBuffer.putInt(current.keyLength);
+    kvBuffer.putInt(current.valueLength);
+    kvBuffer.put(current.keyBuffer.getBytes(), current.keyBuffer.getOffset(),
+        current.keyLength);
+    ByteBufferUtils.copyFromBufferToBuffer(kvBuffer, currentBuffer,
+        current.valueOffset, current.valueLength);
+    if (current.tagsLength > 0) {
+      // Put short as unsigned
+      kvBuffer.put((byte) (current.tagsLength >> 8 & 0xff));
+      kvBuffer.put((byte) (current.tagsLength & 0xff));
+      if (current.tagsOffset != -1) {
+        ByteBufferUtils.copyFromBufferToBuffer(kvBuffer, currentBuffer,
+            current.tagsOffset, current.tagsLength);
+      }
+    }
+    if (includesMvcc()) {
+      ByteBufferUtils.writeVLong(kvBuffer, current.getSequenceId());
+    }
+    kvBuffer.rewind();
+    return kvBuffer;
+  }
+
+  protected ByteBuffer createKVBuffer() {
+    int kvBufSize = (int) KeyValue.getKeyValueDataStructureSize(
+        current.keyLength, current.valueLength, current.tagsLength);
+    if (includesMvcc()) {
+      kvBufSize += WritableUtils.getVIntSize(current.getSequenceId());
+    }
+    ByteBuffer kvBuffer = ByteBuffer.allocate(kvBufSize);
+    return kvBuffer;
+  }
+
+  @Override
+  public Cell getKeyValue() {
+    return current.toCell();
+  }
+
+  @Override
+  public void rewind() {
+    currentBuffer.rewind();
+    decodeFirst();
+  }
+
+  @Override
+  public boolean next() {
+    if (!currentBuffer.hasRemaining()) {
+      return false;
+    }
+    decodeNext();
+    previous.invalidate();
+    return true;
+  }
+
+  @Override
+  public int seekToKeyInBlock(byte[] key, int offset, int length,
+      boolean seekBefore) {
+    return seekToKeyInBlock(new KeyValue.KeyOnlyKeyValue(key, offset, length),
+        seekBefore);
+  }
+
+  private int binarySearch(Cell seekCell, boolean seekBefore) {
+    int low = 0;
+    int high = rowNumber - 1;
+    int mid = (low + high) >>> 1;
+    int comp = 0;
+    SimpleMutableByteRange row = new SimpleMutableByteRange();
+    while (low <= high) {
+      mid = (low + high) >>> 1;
+      getRow(mid, row);
+      comp = comparator.compareRows(row.getBytes(), row.getOffset(),
+          row.getLength(), seekCell.getRowArray(), seekCell.getRowOffset(),
+          seekCell.getRowLength());
+      if (comp < 0) {
+        low = mid + 1;
+      } else if (comp > 0) {
+        high = mid - 1;
+      } else {
+        // key found
+        if (seekBefore) {
+          return mid - 1;
+        } else {
+          return mid;
+        }
+      }
+    }
+    // key not found.
+    if (comp > 0) {
+      return mid - 1;
+    } else {
+      return mid;
+    }
+  }
+
+  private void getRow(int index, SimpleMutableByteRange row) {
+    int offset = Bytes.toIntUnsafe(rowOffsets.array(), rowOffsets.arrayOffset()
+        + (index << 2)); // index * Bytes.SIZEOF_INT
+    int position = currentBuffer.arrayOffset() + offset + Bytes.SIZEOF_LONG;
+    short rowLen = Bytes.toShortUnsafe(currentBuffer.array(), position);
+    row.set(currentBuffer.array(), position + Bytes.SIZEOF_SHORT, rowLen);
+  }
+
+  @Override
+  public int seekToKeyInBlock(Cell seekCell, boolean seekBefore) {
+    previous.invalidate();
+    int index = binarySearch(seekCell, seekBefore);
+    if (index < 0) {
+      return HConstants.INDEX_KEY_MAGIC; // using optimized index key
+    } else {
+      int offset = Bytes.toIntUnsafe(rowOffsets.array(),
+          rowOffsets.arrayOffset() + (index << 2));
+      if (offset != 0) {
+        decodeAtPosition(offset);
+      }
+    }
+    do {
+      int comp;
+      comp = comparator.compareOnlyKeyPortion(seekCell, current.currentKey);
+      if (comp == 0) { // exact match
+        if (seekBefore) {
+          if (!previous.isValid()) {
+            // The caller (seekBefore) has to ensure that we are not at the
+            // first key in the block.
+            throw new IllegalStateException("Cannot seekBefore if "
+                + "positioned at the first key in the block: key="
+                + Bytes.toStringBinary(seekCell.getRowArray()));
+          }
+          moveToPrevious();
+          return 1;
+        }
+        return 0;
+      }
+
+      if (comp < 0) { // already too large, check previous
+        if (previous.isValid()) {
+          moveToPrevious();
+        } else {
+          return HConstants.INDEX_KEY_MAGIC; // using optimized index key
+        }
+        return 1;
+      }
+
+      // move to next, if more data is available
+      if (currentBuffer.hasRemaining()) {
+        previous.copyFromNext(current);
+        decodeNext();
+      } else {
+        break;
+      }
+    } while (true);
+
+    // we hit the end of the block, not an exact match
+    return 1;
+  }
+
+  private void moveToPrevious() {
+    if (!previous.isValid()) {
+      throw new IllegalStateException(
+          "Can move back only once and not in first key in the block.");
+    }
+
+    SeekerState tmp = previous;
+    previous = current;
+    current = tmp;
+
+    // move after last key value
+    currentBuffer.position(current.nextKvOffset);
+    previous.invalidate();
+  }
+
+  @Override
+  public int compareKey(KVComparator comparator, byte[] key, int offset,
+      int length) {
+    return comparator.compareFlatKey(key, offset, length,
+        current.keyBuffer.getBytes(), current.keyBuffer.getOffset(),
+        current.keyBuffer.getLength());
+  }
+
+  @Override
+  public int compareKey(KVComparator comparator, Cell key) {
+    return comparator.compareOnlyKeyPortion(key, new KeyValue.KeyOnlyKeyValue(
+        current.keyBuffer.getBytes(), current.keyBuffer.getOffset(),
+        current.keyBuffer.getLength()));
+  }
+
+  protected void decodeFirst() {
+    decodeNext();
+    previous.invalidate();
+  }
+
+  protected void decodeAtPosition(int position) {
+    currentBuffer.position(position);
+    decodeNext();
+    previous.invalidate();
+  }
+
+  protected void decodeNext() {
+    current.startOffset = currentBuffer.position();
+    int p = currentBuffer.position() + currentBuffer.arrayOffset();
+    long ll = Bytes.toLong(currentBuffer.array(), p);
+    // Read top half as an int of key length and bottom int as value length
+    current.keyLength = (int) (ll >> Integer.SIZE);
+    current.valueLength = (int) (Bytes.MASK_FOR_LOWER_INT_IN_LONG ^ ll);
+    ByteBufferUtils.skip(currentBuffer, Bytes.SIZEOF_LONG);
+    // key part
+    current.keyBuffer.set(currentBuffer.array(), currentBuffer.arrayOffset()
+        + currentBuffer.position(), current.keyLength);
+    ByteBufferUtils.skip(currentBuffer, current.keyLength);
+    // value part
+    current.valueOffset = currentBuffer.position();
+    ByteBufferUtils.skip(currentBuffer, current.valueLength);
+    if (includesTags()) {
+      decodeTags();
+    }
+    if (includesMvcc()) {
+      current.memstoreTS = ByteBufferUtils.readVLong(currentBuffer);
+    } else {
+      current.memstoreTS = 0;
+    }
+    current.nextKvOffset = currentBuffer.position();
+    current.setKey(current.keyBuffer.getBytes(), current.keyBuffer.getOffset(),
+        current.keyBuffer.getLength());
+  }
+
+  protected boolean includesMvcc() {
+    return this.decodingCtx.getHFileContext().isIncludesMvcc();
+  }
+
+  protected boolean includesTags() {
+    return this.decodingCtx.getHFileContext().isIncludesTags();
+  }
+
+  protected void decodeTags() {
+    current.tagsLength = currentBuffer.getShort();
+    current.tagsOffset = currentBuffer.position();
+    ByteBufferUtils.skip(currentBuffer, current.tagsLength);
+  }
+
+  protected class SeekerState {
+    /**
+     * The size of a (key length, value length) tuple that prefixes each entry
+     * in a data block.
+     */
+    public final static int KEY_VALUE_LEN_SIZE = 2 * Bytes.SIZEOF_INT;
+
+    protected ByteBuffer currentBuffer;
+    protected int startOffset = -1;
+    protected int valueOffset = -1;
+    protected int keyLength;
+    protected int valueLength;
+    protected int tagsLength = 0;
+    protected int tagsOffset = -1;
+
+    protected SimpleMutableByteRange keyBuffer = new SimpleMutableByteRange();
+    protected long memstoreTS;
+    protected int nextKvOffset;
+    protected KeyValue.KeyOnlyKeyValue currentKey = new KeyValue.KeyOnlyKeyValue();
+
+    protected boolean isValid() {
+      return valueOffset != -1;
+    }
+
+    protected void invalidate() {
+      valueOffset = -1;
+      currentKey = new KeyValue.KeyOnlyKeyValue();
+      currentBuffer = null;
+    }
+
+    protected void setKey(byte[] key, int offset, int length) {
+      currentKey.setKey(key, offset, length);
+    }
+
+    protected long getSequenceId() {
+      return memstoreTS;
+    }
+
+    /**
+     * Copy the state from the next one into this instance (the previous state
+     * placeholder). Used to save the previous state when we are advancing the
+     * seeker to the next key/value.
+     */
+    protected void copyFromNext(SeekerState nextState) {
+      keyBuffer.set(nextState.keyBuffer.getBytes(),
+          nextState.keyBuffer.getOffset(), nextState.keyBuffer.getLength());
+      currentKey.setKey(nextState.keyBuffer.getBytes(),
+          nextState.keyBuffer.getOffset(), nextState.keyBuffer.getLength());
+
+      startOffset = nextState.startOffset;
+      valueOffset = nextState.valueOffset;
+      keyLength = nextState.keyLength;
+      valueLength = nextState.valueLength;
+      nextKvOffset = nextState.nextKvOffset;
+      memstoreTS = nextState.memstoreTS;
+      currentBuffer = nextState.currentBuffer;
+      tagsOffset = nextState.tagsOffset;
+      tagsLength = nextState.tagsLength;
+    }
+
+    @Override
+    public String toString() {
+      return CellUtil.getCellKeyAsString(toCell());
+    }
+
+    protected int getCellBufSize() {
+      int kvBufSize = KEY_VALUE_LEN_SIZE + keyLength + valueLength;
+      if (includesTags()) {
+        kvBufSize += Bytes.SIZEOF_SHORT + tagsLength;
+      }
+      return kvBufSize;
+    }
+
+    protected Cell formNoTagsKeyValue() {
+      NoTagsKeyValue ret = new NoTagsKeyValue(currentBuffer.array(),
+          currentBuffer.arrayOffset() + startOffset, getCellBufSize());
+      if (includesMvcc()) {
+        ret.setSequenceId(memstoreTS);
+      }
+      return ret;
+    }
+
+    public Cell toCell() {
+      if (tagsOffset > 0) {
+        KeyValue ret = new KeyValue(currentBuffer.array(),
+            currentBuffer.arrayOffset() + startOffset, getCellBufSize());
+        if (includesMvcc()) {
+          ret.setSequenceId(memstoreTS);
+        }
+        return ret;
+      } else {
+        return formNoTagsKeyValue();
+      }
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/c899897b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/encoding/TestSeekToBlockWithEncoders.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/encoding/TestSeekToBlockWithEncoders.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/encoding/TestSeekToBlockWithEncoders.java
index 914a37b..5f8c1ef 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/encoding/TestSeekToBlockWithEncoders.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/encoding/TestSeekToBlockWithEncoders.java
@@ -83,7 +83,7 @@ public class TestSeekToBlockWithEncoders {
     KeyValue kv4 = new KeyValue(Bytes.toBytes("aad"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
         Bytes.toBytes("val"));
     sampleKv.add(kv4);
-    KeyValue kv5 = new KeyValue(Bytes.toBytes("aaaad"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
+    KeyValue kv5 = new KeyValue(Bytes.toBytes("aaddd"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
         Bytes.toBytes("val"));
     sampleKv.add(kv5);
     KeyValue toSeek = new KeyValue(Bytes.toBytes("aaaa"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
@@ -106,7 +106,7 @@ public class TestSeekToBlockWithEncoders {
     KeyValue kv3 = new KeyValue(Bytes.toBytes("aac"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
         Bytes.toBytes("val"));
     sampleKv.add(kv3);
-    KeyValue kv4 = new KeyValue(Bytes.toBytes("aaae"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
+    KeyValue kv4 = new KeyValue(Bytes.toBytes("aade"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),
         Bytes.toBytes("val"));
     sampleKv.add(kv4);
     KeyValue kv5 = new KeyValue(Bytes.toBytes("bbbcd"), Bytes.toBytes("f1"), Bytes.toBytes("q1"),


Mime
View raw message