Java 类org.apache.hadoop.hbase.Cell 实例源码

private List<Cell> getScanResult(byte[] startRow, byte[] stopRow, HTable ht) throws IOException {
  Scan scan = new Scan();
  scan.setMaxVersions();
  if(!Bytes.toString(startRow).isEmpty()) {
    scan.setStartRow(startRow);
  }
  if(!Bytes.toString(stopRow).isEmpty()) {
    scan.setStopRow(stopRow);
  }
  ResultScanner scanner = ht.getScanner(scan);
  List<Cell> kvList = new ArrayList<Cell>();
  Result r;
  while ((r = scanner.next()) != null) {
    for (Cell kv : r.listCells()) {
      kvList.add(kv);
    }
  }
  return kvList;
}

@Override
public int encode(Cell cell, HFileBlockEncodingContext encodingCtx, DataOutputStream out)
    throws IOException {
  int klength = KeyValueUtil.keyLength(cell);
  int vlength = cell.getValueLength();

  out.writeInt(klength);
  out.writeInt(vlength);
  CellUtil.writeFlatKey(cell, out);
  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);
    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());
  }
  return encodedKvSize;
}

@Test
public void testMultiRowRangeFilterWithoutRangeOverlap() throws IOException {
  tableName = Bytes.toBytes("testMultiRowRangeFilterWithoutRangeOverlap");
  HTable ht = TEST_UTIL.createTable(tableName, family, Integer.MAX_VALUE);
  generateRows(numRows, ht, family, qf, value);

  Scan scan = new Scan();
  scan.setMaxVersions();

  List<RowRange> ranges = new ArrayList<RowRange>();
  ranges.add(new RowRange(Bytes.toBytes(30), true, Bytes.toBytes(40), false));
  ranges.add(new RowRange(Bytes.toBytes(10), true, Bytes.toBytes(20), false));
  ranges.add(new RowRange(Bytes.toBytes(60), true, Bytes.toBytes(70), false));

  MultiRowRangeFilter filter = new MultiRowRangeFilter(ranges);
  scan.setFilter(filter);
  int resultsSize = getResultsSize(ht, scan);
  LOG.info("found " + resultsSize + " results");
  List<Cell> results1 = getScanResult(Bytes.toBytes(10), Bytes.toBytes(20), ht);
  List<Cell> results2 = getScanResult(Bytes.toBytes(30), Bytes.toBytes(40), ht);
  List<Cell> results3 = getScanResult(Bytes.toBytes(60), Bytes.toBytes(70), ht);

  assertEquals(results1.size() + results2.size() + results3.size(), resultsSize);

  ht.close();
}

private void verifyExpectedCounts(Table table, Scan scan, int expectedRowCount,
    int expectedCellCount) throws Exception {
  ResultScanner scanner = table.getScanner(scan);

  int rowCount = 0;
  int cellCount = 0;
  Result r = null;
  while ((r = scanner.next()) != null) {
    rowCount++;
    for (Cell c : r.rawCells()) {
      cellCount++;
    }
  }

  assertTrue("Expected row count: " + expectedRowCount + " Actual row count: " + rowCount,
    expectedRowCount == rowCount);
  assertTrue("Expected cell count: " + expectedCellCount + " Actual cell count: " + cellCount,
    expectedCellCount == cellCount);
  scanner.close();
}

public void testDeletedRowThenGoodRow() throws IOException {
  KeyValue [] kvs = new KeyValue [] {
      KeyValueTestUtil.create("R1", "cf", "a", 1, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "a", 1, KeyValue.Type.Delete, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "a", 20, KeyValue.Type.Put, "dont-care")
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scanSpec = new Scan(Bytes.toBytes("R1"));
  StoreScanner scan = new StoreScanner(scanSpec, scanInfo, scanType,
      getCols("a"), scanners);

  List<Cell> results = new ArrayList<Cell>();
  assertEquals(true, scan.next(results));
  assertEquals(0, results.size());

  assertEquals(true, scan.next(results));
  assertEquals(1, results.size());
  assertEquals(kvs[2], results.get(0));

  assertEquals(false, scan.next(results));
}

private void printKV(Cell keyValue) {
  StringBuilder sb = new StringBuilder();
  sb.append("rowkey=" + Bytes.toStringBinary(keyValue.getRow()));
  int i = 0;
  int[] arr = MDUtils.bitwiseUnzip(keyValue.getRow(), 3);
  sb.append(", indicating=");
  for (Map.Entry<byte[], TreeSet<byte[]>> entry : relation.getIndexFamilyMap().entrySet()) {
    for (byte[] qualifer : entry.getValue()) {
      sb.append("[").append(Bytes.toString(entry.getKey())).append(":")
          .append(Bytes.toString(qualifer)).append("]=").append(arr[i]).append(",");
      ++i;
    }
  }
  sb.append(", rawRowkey=" + Bytes.toInt(keyValue.getQualifier()));
  System.out.println(sb.toString());
}

protected List<List<Cell>> getExistingLabelsWithAuths() throws IOException {
  Scan scan = new Scan();
  RegionScanner scanner = labelsRegion.getScanner(scan);
  List<List<Cell>> existingLabels = new ArrayList<List<Cell>>();
  try {
    while (true) {
      List<Cell> cells = new ArrayList<Cell>();
      scanner.next(cells);
      if (cells.isEmpty()) {
        break;
      }
      existingLabels.add(cells);
    }
  } finally {
    scanner.close();
  }
  return existingLabels;
}

private boolean walkForwardInSingleRow(final HFileScanner scanner, final KeyValue firstOnRow,
    final GetClosestRowBeforeTracker state) throws IOException {
  boolean foundCandidate = false;
  do {
    Cell kv = scanner.getKeyValue();
    // If we are not in the row, skip.
    if (this.comparator.compareRows(kv, firstOnRow) < 0) continue;
    // Did we go beyond the target row? If so break.
    if (state.isTooFar(kv, firstOnRow)) break;
    if (state.isExpired(kv)) {
      continue;
    }
    // If we added something, this row is a contender. break.
    if (state.handle(kv)) {
      foundCandidate = true;
      break;
    }
  } while (scanner.next());
  return foundCandidate;
}

/**
 * Method to account for the size of retained cells and retained data blocks.
 * @return an object that represents the last referenced block from this response.
 */
Object addSize(RpcCallContext context, Result r, Object lastBlock) {
  if (context != null && !r.isEmpty()) {
    for (Cell c : r.rawCells()) {
      context.incrementResponseCellSize(CellUtil.estimatedHeapSizeOf(c));
      // We're using the last block being the same as the current block as
      // a proxy for pointing to a new block. This won't be exact.
      // If there are multiple gets that bounce back and forth
      // Then it's possible that this will over count the size of
      // referenced blocks. However it's better to over count and
      // use two RPC's than to OOME the RegionServer.
      byte[] valueArray = c.getValueArray();
      if (valueArray != lastBlock) {
        context.incrementResponseBlockSize(valueArray.length);
        lastBlock = valueArray;
      }
    }
  }
  return lastBlock;
}

@Override
public long append(HTableDescriptor htd, HRegionInfo info, WALKey key, WALEdit edits,
                   boolean inMemstore) {
  if (!this.listeners.isEmpty()) {
    final long start = System.nanoTime();
    long len = 0;
    for (Cell cell : edits.getCells()) {
      len += CellUtil.estimatedSerializedSizeOf(cell);
    }
    final long elapsed = (System.nanoTime() - start)/1000000l;
    for (WALActionsListener listener : this.listeners) {
      listener.postAppend(len, elapsed);
    }
  }
  return -1;
}

@Override
public void write(DataOutput out) throws IOException {
  LOG.warn("WALEdit is being serialized to writable - only expected in test code");
  out.writeInt(VERSION_2);
  out.writeInt(cells.size());
  // We interleave the two lists for code simplicity
  for (Cell cell : cells) {
    // This is not used in any of the core code flows so it is just fine to convert to KV
    KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
    if (compressionContext != null) {
      KeyValueCompression.writeKV(out, kv, compressionContext);
    } else{
      KeyValue.write(kv, out);
    }
  }
  if (scopes == null) {
    out.writeInt(0);
  } else {
    out.writeInt(scopes.size());
    for (byte[] key : scopes.keySet()) {
      Bytes.writeByteArray(out, key);
      out.writeInt(scopes.get(key));
    }
  }
}

@Override
public void preGetOp(final ObserverContext<RegionCoprocessorEnvironment> e,
                     final Get get, final List<Cell> results) throws IOException {

  if (e.getEnvironment().getRegion().getRegionInfo().getReplicaId() == 0) {
    CountDownLatch latch = cdl.get();
    try {
      if (sleepTime.get() > 0) {
        LOG.info("Sleeping for " + sleepTime.get() + " ms");
        Thread.sleep(sleepTime.get());
      } else if (latch.getCount() > 0) {
        LOG.info("Waiting for the counterCountDownLatch");
        latch.await(2, TimeUnit.MINUTES); // To help the tests to finish.
        if (latch.getCount() > 0) {
          throw new RuntimeException("Can't wait more");
        }
      }
    } catch (InterruptedException e1) {
      LOG.error(e1);
    }
  } else {
    LOG.info("We're not the primary replicas.");
  }
}

@Override
public WALEntryFilter getWALEntryfilter() {
  return new ChainWALEntryFilter(super.getWALEntryfilter(), new WALEntryFilter() {
    @Override
    public Entry filter(Entry entry) {
      ArrayList<Cell> cells = entry.getEdit().getCells();
      int size = cells.size();
      for (int i = size-1; i >= 0; i--) {
        Cell cell = cells.get(i);
        if (!Bytes.equals(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
          row, 0, row.length)) {
          cells.remove(i);
        }
      }
      return entry;
    }
  });
}

private void scanRow(final Result result, final RowKeyBuilder simpleRowKeyBuilder, final RowKey rowKey,
                     final StatisticType statsType, EventStoreTimeIntervalEnum interval) throws IOException {
    final CellScanner cellScanner = result.cellScanner();
    while (cellScanner.advance()) {
        final Cell cell = cellScanner.current();

        // get the column qualifier
        final byte[] bTimeQualifier = new byte[cell.getQualifierLength()];
        System.arraycopy(cell.getQualifierArray(), cell.getQualifierOffset(), bTimeQualifier, 0,
                cell.getQualifierLength());

        // convert this into a true time, albeit rounded to the column
        // interval granularity
        final long columnIntervalNo = Bytes.toInt(bTimeQualifier);
        final long columnIntervalSize = interval.columnInterval();
        final long columnTimeComponentMillis = columnIntervalNo * columnIntervalSize;
        final long rowKeyPartialTimeMillis = simpleRowKeyBuilder.getPartialTimestamp(rowKey);
        final long fullTimestamp = rowKeyPartialTimeMillis + columnTimeComponentMillis;

        LOGGER.debug("Col: [" + ByteArrayUtils.byteArrayToHex(bTimeQualifier) + "] - ["
                + Bytes.toInt(bTimeQualifier) + "] - [" + fullTimestamp + "] - ["
                + DateUtil.createNormalDateTimeString(fullTimestamp) + "]");

        final byte[] bValue = new byte[cell.getValueLength()];
        System.arraycopy(cell.getValueArray(), cell.getValueOffset(), bValue, 0, cell.getValueLength());

        switch (statsType) {
            case VALUE:
                final ValueCellValue cellValue = new ValueCellValue(bValue);

                LOGGER.debug("Val: " + cellValue);
                break;
            case COUNT:
                LOGGER.debug("Val: " + Bytes.toLong(bValue));
                break;
        }

    }
}

@Override
public ReturnCode filterKeyValue(Cell v) {
  if (filterOutRow) {
    return ReturnCode.NEXT_ROW;
  }
  return ReturnCode.INCLUDE;
}

/**
 * Test written as a verifier for HBASE-7051, CheckAndPut should properly read
 * MVCC.
 *
 * Moved into TestAtomicOperation from its original location, TestHBase7051
 */
@Test
public void testPutAndCheckAndPutInParallel() throws Exception {

  final String tableName = "testPutAndCheckAndPut";
  Configuration conf = TEST_UTIL.getConfiguration();
  conf.setClass(HConstants.REGION_IMPL, MockHRegion.class, HeapSize.class);
  final MockHRegion region = (MockHRegion) TEST_UTIL.createLocalHRegion(Bytes.toBytes(tableName),
      null, null, tableName, conf, false, Durability.SYNC_WAL, null, Bytes.toBytes(family));

  Put[] puts = new Put[1];
  Put put = new Put(Bytes.toBytes("r1"));
  put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("10"));
  puts[0] = put;

  region.batchMutate(puts, HConstants.NO_NONCE, HConstants.NO_NONCE);
  MultithreadedTestUtil.TestContext ctx =
    new MultithreadedTestUtil.TestContext(conf);
  ctx.addThread(new PutThread(ctx, region));
  ctx.addThread(new CheckAndPutThread(ctx, region));
  ctx.startThreads();
  while (testStep != TestStep.CHECKANDPUT_COMPLETED) {
    Thread.sleep(100);
  }
  ctx.stop();
  Scan s = new Scan();
  RegionScanner scanner = region.getScanner(s);
  List<Cell> results = new ArrayList<Cell>();
  ScannerContext scannerContext = ScannerContext.newBuilder().setBatchLimit(2).build();
  scanner.next(results, scannerContext);
  for (Cell keyValue : results) {
    assertEquals("50",Bytes.toString(CellUtil.cloneValue(keyValue)));
  }
}

private Cell getLowest(Cell first, Cell second) {
  if (first == null && second == null) {
    return null;
  }
  if (first != null && second != null) {
    int compare = comparator.compare(first, second);
    return (compare <= 0 ? first : second);
  }
  return (first != null ? first : second);
}

protected void resetScannerStack(Cell lastTopKey) throws IOException {
  if (heap != null) {
    throw new RuntimeException("StoreScanner.reseek run on an existing heap!");
  }

  /* When we have the scan object, should we not pass it to getScanners()
   * to get a limited set of scanners? We did so in the constructor and we
   * could have done it now by storing the scan object from the constructor */
  List<KeyValueScanner> scanners = getScannersNoCompaction();

  // Seek all scanners to the initial key
  seekScanners(scanners, lastTopKey, false, parallelSeekEnabled);

  // Combine all seeked scanners with a heap
  resetKVHeap(scanners, store.getComparator());

  // Reset the state of the Query Matcher and set to top row.
  // Only reset and call setRow if the row changes; avoids confusing the
  // query matcher if scanning intra-row.
  Cell kv = heap.peek();
  if (kv == null) {
    kv = lastTopKey;
  }
  byte[] row = kv.getRowArray();
  int offset = kv.getRowOffset();
  short length = kv.getRowLength();
  if ((matcher.row == null) || !Bytes
      .equals(row, offset, length, matcher.row, matcher.rowOffset, matcher.rowLength)) {
    this.countPerRow = 0;
    matcher.reset();
    matcher.setRow(row, offset, length);
  }
}

/**
 * Delete the specified version of the specified column.
 * @param family family name
 * @param qualifier column qualifier
 * @param timestamp version timestamp
 * @return this for invocation chaining
 */
public Delete addColumn(byte [] family, byte [] qualifier, long timestamp) {
  if (timestamp < 0) {
    throw new IllegalArgumentException("Timestamp cannot be negative. ts=" + timestamp);
  }
  List<Cell> list = familyMap.get(family);
  if(list == null) {
    list = new ArrayList<Cell>();
  }
  KeyValue kv = new KeyValue(this.row, family, qualifier, timestamp, KeyValue.Type.Delete);
  list.add(kv);
  familyMap.put(family, list);
  return this;
}

@Override
public int internalEncode(Cell cell, HFileBlockDefaultEncodingContext encodingContext,
    DataOutputStream out) throws IOException {
  int klength = KeyValueUtil.keyLength(cell);
  int vlength = cell.getValueLength();
  EncodingState state = encodingContext.getEncodingState();
  if (state.prevCell == null) {
    // copy the key, there is no common prefix with none
    ByteBufferUtils.putCompressedInt(out, klength);
    ByteBufferUtils.putCompressedInt(out, vlength);
    ByteBufferUtils.putCompressedInt(out, 0);
    CellUtil.writeFlatKey(cell, out);
  } else {
    // find a common prefix and skip it
    int common = CellUtil.findCommonPrefixInFlatKey(cell, state.prevCell, true, true);
    ByteBufferUtils.putCompressedInt(out, klength - common);
    ByteBufferUtils.putCompressedInt(out, vlength);
    ByteBufferUtils.putCompressedInt(out, common);
    writeKeyExcludingCommon(cell, common, out);
  }
  // Write the value part
  out.write(cell.getValueArray(), cell.getValueOffset(), vlength);
  int size = klength + vlength + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE;
  size += afterEncodingKeyValue(cell, out, encodingContext);
  state.prevCell = cell;
  return size;
}

@Override
public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e, final Put put,
    final WALEdit edit, final Durability durability) throws IOException {
  byte[] attribute = put.getAttribute("visibility");
  byte[] cf = null;
  List<Cell> updatedCells = new ArrayList<Cell>();
  if (attribute != null) {
    for (List<? extends Cell> edits : put.getFamilyCellMap().values()) {
      for (Cell cell : edits) {
        KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
        if (cf == null) {
          cf = kv.getFamily();
        }
        Tag tag = new Tag(TAG_TYPE, attribute);
        List<Tag> tagList = new ArrayList<Tag>();
        tagList.add(tag);

        KeyValue newKV = new KeyValue(kv.getRow(), 0, kv.getRowLength(), kv.getFamily(), 0,
            kv.getFamilyLength(), kv.getQualifier(), 0, kv.getQualifierLength(),
            kv.getTimestamp(), KeyValue.Type.codeToType(kv.getType()), kv.getValue(), 0,
            kv.getValueLength(), tagList);
        ((List<Cell>) updatedCells).add(newKV);
      }
    }
    put.getFamilyCellMap().remove(cf);
    // Update the family map
    put.getFamilyCellMap().put(cf, updatedCells);
  }
}

private void scanAll(Result[] next) throws IOException {
  CellScanner cellScanner = next[0].cellScanner();
  cellScanner.advance();
  Cell current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 127l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 126l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 125l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 124l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 123l);
  cellScanner = next[1].cellScanner();
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row2, 0, row2.length));
}

/** Tests that the rpc scheduler is called when requests arrive. */
@Test
public void testRpcScheduler() throws IOException, InterruptedException {
  RpcScheduler scheduler = spy(new FifoRpcScheduler(CONF, 1));
  RpcServer rpcServer = new TestRpcServer(scheduler);
  verify(scheduler).init((RpcScheduler.Context) anyObject());
  AbstractRpcClient client = createRpcClient(CONF);
  try {
    rpcServer.start();
    verify(scheduler).start();
    MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
    EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    for (int i = 0; i < 10; i++) {
      client.call(new PayloadCarryingRpcController(
          CellUtil.createCellScanner(ImmutableList.<Cell> of(CELL))), md, param,
          md.getOutputType().toProto(), User.getCurrent(), address,
          new MetricsConnection.CallStats());
    }
    verify(scheduler, times(10)).dispatch((CallRunner) anyObject());
  } finally {
    rpcServer.stop();
    verify(scheduler).stop();
  }
}

/**
 * Count the number of different row keys in the given edit because of mini-batching. We assume
 * that there's at least one Cell in the WALEdit.
 * @param edit edit to count row keys from
 * @return number of different row keys
 */
private int countDistinctRowKeys(WALEdit edit) {
  List<Cell> cells = edit.getCells();
  int distinctRowKeys = 1;
  Cell lastCell = cells.get(0);
  for (int i = 0; i < edit.size(); i++) {
    if (!CellUtil.matchingRow(cells.get(i), lastCell)) {
      distinctRowKeys++;
    }
  }
  return distinctRowKeys;
}

/**
 * Add the specified column and value, with the specified timestamp as
 * its version to this Put operation.
 * @param family family name
 * @param qualifier column qualifier
 * @param ts version timestamp
 * @param value column value
 * @return this
 */
public Put addColumn(byte [] family, byte [] qualifier, long ts, byte [] value) {
  if (ts < 0) {
    throw new IllegalArgumentException("Timestamp cannot be negative. ts=" + ts);
  }
  List<Cell> list = getCellList(family);
  KeyValue kv = createPutKeyValue(family, qualifier, ts, value);
  list.add(kv);
  familyMap.put(CellUtil.cloneFamily(kv), list);
  return this;
}

/**************** helper **************************/

  protected void assertKeyAndValueEqual(Cell expected, Cell actual) {
    // assert keys are equal (doesn't compare values)
    Assert.assertEquals(expected, actual);
    if (includeMemstoreTS) {
      Assert.assertEquals(expected.getMvccVersion(), actual.getMvccVersion());
    }
    // assert values equal
    Assert.assertTrue(Bytes.equals(expected.getValueArray(), expected.getValueOffset(),
      expected.getValueLength(), actual.getValueArray(), actual.getValueOffset(),
      actual.getValueLength()));
  }

private ArrayList<Cell> init(RegionScanner regionScanner, ScanRange primaryRange,
    IndexPutParser parse) throws IOException {
  ArrayList<Cell> ret = new ArrayList<>();
  while (true) {
    List<Cell> oneRow = new ArrayList<>();
    if (!regionScanner.nextRaw(oneRow)) break;
    processOneRow(ret, oneRow, parse);
    //      processOneRow(ret, oneRow, primaryRange, relation);
  }
  Collections.sort(ret, KeyValue.COMPARATOR);
  return ret;
}

/**
 * Basic test to verify utility methods in {@link PBType} and delegation to protobuf works.
 */
@Test
public void testRoundTrip() {
  final Cell cell = new KeyValue(Bytes.toBytes("row"), Bytes.toBytes("fam"),
    Bytes.toBytes("qual"), Bytes.toBytes("val"));
  CellProtos.Cell c = ProtobufUtil.toCell(cell), decoded;
  PositionedByteRange pbr = new SimplePositionedByteRange(c.getSerializedSize());
  pbr.setPosition(0);
  int encodedLength = CODEC.encode(pbr, c);
  pbr.setPosition(0);
  decoded = CODEC.decode(pbr);
  assertEquals(encodedLength, pbr.getPosition());
  assertTrue(CellComparator.equals(cell, ProtobufUtil.toCell(decoded)));
}

@Override
public ReturnCode filterKeyValue(Cell cell) throws IOException {
  List<Tag> putVisTags = new ArrayList<Tag>();
  Byte putCellVisTagsFormat = VisibilityUtils.extractVisibilityTags(cell, putVisTags);
  boolean matchFound = VisibilityLabelServiceManager
      .getInstance().getVisibilityLabelService()
      .matchVisibility(putVisTags, putCellVisTagsFormat, deleteCellVisTags,
          deleteCellVisTagsFormat);
  return matchFound ? ReturnCode.INCLUDE : ReturnCode.SKIP;
}

private boolean matchingColumn(Cell keyValue, byte[] family, byte[] qualifier) {
  if (!Bytes.equals(family, keyValue.getFamilyArray())) {
    return false;
  }
  if (qualifier == null || qualifier.length == 0) {
    if (keyValue.getQualifierLength() == 0) {
      return true;
    }
    return false;
  }
  return Bytes.equals(qualifier, keyValue.getQualifierArray());
}

protected int seekToOrBeforeUsingPositionAtOrAfter(Cell kv, boolean seekBefore) {
  // should probably switch this to use the seekForwardToOrBefore method
  CellScannerPosition position = ptSearcher.seekForwardToOrAfter(kv);

  if (CellScannerPosition.AT == position) {
    if (seekBefore) {
      ptSearcher.previous();
      return 1;
    }
    return 0;

  }

  if (CellScannerPosition.AFTER == position) {
    if (!ptSearcher.isBeforeFirst()) {
      ptSearcher.previous();
    }
    return 1;
  }

  if (position == CellScannerPosition.AFTER_LAST) {
    if (seekBefore) {
      ptSearcher.previous();
    }
    return 1;
  }

  throw new RuntimeException("unexpected CellScannerPosition:" + position);
}

public Cell next() {
  if (cache != null) {
    Cell kv = cache;
    cache = null;
    return kv;
  }
  if (valuesI == null) {
    return null;
  }
  try {
    return valuesI.next();
  } catch (NoSuchElementException e) {
    return null;
  }
}

@Override
public ReturnCode filterKeyValue(Cell v) {
  if (doCompare(this.compareOp, this.comparator, v.getValueArray(),
      v.getValueOffset(), v.getValueLength())) {
    return ReturnCode.SKIP;
  }
  return ReturnCode.INCLUDE;
}

@Override
public List<String> getGroupAuths(String[] groups, boolean systemCall)
    throws IOException {
  assert (labelsRegion != null || systemCall);
  if (systemCall || labelsRegion == null) {
    return this.labelsCache.getGroupAuths(groups);
  }
  Scan s = new Scan();
  if (groups != null && groups.length > 0) {
    for (String group : groups) {
      s.addColumn(LABELS_TABLE_FAMILY, Bytes.toBytes(AuthUtil.toGroupEntry(group)));
    }
  }
  Filter filter = VisibilityUtils.createVisibilityLabelFilter(this.labelsRegion,
      new Authorizations(SYSTEM_LABEL));
  s.setFilter(filter);
  Set<String> auths = new HashSet<String>();
  RegionScanner scanner = this.labelsRegion.getScanner(s);
  try {
    List<Cell> results = new ArrayList<Cell>(1);
    while (true) {
      scanner.next(results);
      if (results.isEmpty()) break;
      Cell cell = results.get(0);
      int ordinal = Bytes.toInt(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
      String label = this.labelsCache.getLabel(ordinal);
      if (label != null) {
        auths.add(label);
      }
      results.clear();
    }
  } finally {
    scanner.close();
  }
  return new ArrayList<String>(auths);
}

/**
 * Searches for the latest value for the specified column.
 *
 * @param kvs the array to search
 * @param family family name
 * @param foffset family offset
 * @param flength family length
 * @param qualifier column qualifier
 * @param qoffset qualifier offset
 * @param qlength qualifier length
 *
 * @return the index where the value was found, or -1 otherwise
 */
protected int binarySearch(final Cell [] kvs,
    final byte [] family, final int foffset, final int flength,
    final byte [] qualifier, final int qoffset, final int qlength) {

  double keyValueSize = (double)
      KeyValue.getKeyValueDataStructureSize(kvs[0].getRowLength(), flength, qlength, 0);

  byte[] buffer = localBuffer.get();
  if (buffer == null || keyValueSize > buffer.length) {
    // pad to the smallest multiple of the pad width
    buffer = new byte[(int) Math.ceil(keyValueSize / PAD_WIDTH) * PAD_WIDTH];
    localBuffer.set(buffer);
  }

  Cell searchTerm = KeyValueUtil.createFirstOnRow(buffer, 0,
      kvs[0].getRowArray(), kvs[0].getRowOffset(), kvs[0].getRowLength(),
      family, foffset, flength,
      qualifier, qoffset, qlength);

  // pos === ( -(insertion point) - 1)
  int pos = Arrays.binarySearch(kvs, searchTerm, KeyValue.COMPARATOR);
  // never will exact match
  if (pos < 0) {
    pos = (pos+1) * -1;
    // pos is now insertion point
  }
  if (pos == kvs.length) {
    return -1; // doesn't exist
  }
  return pos;
}

public boolean isMetaEdit() {
  for (Cell cell: cells) {
    if (!isMetaEditFamily(cell)) {
      return false;
    }
  }
  return true;
}

private Cell doTestOfSeekBefore(Path p, FileSystem fs, Reference bottom, KeyValue seekBefore,
                                      CacheConfig cacheConfig)
          throws IOException {
    final HalfStoreFileReader halfreader = new HalfStoreFileReader(fs, p,
            cacheConfig, bottom, TEST_UTIL.getConfiguration());
    halfreader.loadFileInfo();
    final HFileScanner scanner = halfreader.getScanner(false, false);
    scanner.seekBefore(seekBefore);
    return scanner.getKeyValue();
}

@Override
public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e,
    final Put put, final WALEdit edit, final Durability durability)
    throws IOException {
  Map<byte[], List<Cell>> familyMap = put.getFamilyCellMap();
  if (familyMap.containsKey(test)) {
    e.bypass();
  }
}

private void _testBlocksScanned(HTableDescriptor table) throws Exception {
  Region r = createNewHRegion(table, START_KEY, END_KEY, TEST_UTIL.getConfiguration());
  addContent(r, FAMILY, COL);
  r.flush(true);

  CacheStats stats = new CacheConfig(TEST_UTIL.getConfiguration()).getBlockCache().getStats();
  long before = stats.getHitCount() + stats.getMissCount();
  // Do simple test of getting one row only first.
  Scan scan = new Scan(Bytes.toBytes("aaa"), Bytes.toBytes("aaz"));
  scan.addColumn(FAMILY, COL);
  scan.setMaxVersions(1);

  InternalScanner s = r.getScanner(scan);
  List<Cell> results = new ArrayList<Cell>();
  while (s.next(results))
    ;
  s.close();

  int expectResultSize = 'z' - 'a';
  assertEquals(expectResultSize, results.size());

  int kvPerBlock = (int) Math.ceil(BLOCK_SIZE / 
      (double) KeyValueUtil.ensureKeyValue(results.get(0)).getLength());
  Assert.assertEquals(2, kvPerBlock);

  long expectDataBlockRead = (long) Math.ceil(expectResultSize / (double) kvPerBlock);
  long expectIndexBlockRead = expectDataBlockRead;

  assertEquals(expectIndexBlockRead+expectDataBlockRead, stats.getHitCount() + stats.getMissCount() - before);
}

@Override
public void postGetOp(final ObserverContext<RegionCoprocessorEnvironment> c, final Get get,
    final List<Cell> results) {
  RegionCoprocessorEnvironment e = c.getEnvironment();
  assertNotNull(e);
  assertNotNull(e.getRegion());
  assertNotNull(get);
  assertNotNull(results);
  if (e.getRegion().getTableDesc().getTableName().equals(
      TestRegionObserverInterface.TEST_TABLE)) {
    boolean foundA = false;
    boolean foundB = false;
    boolean foundC = false;
    for (Cell kv: results) {
      if (CellUtil.matchingFamily(kv, TestRegionObserverInterface.A)) {
        foundA = true;
      }
      if (CellUtil.matchingFamily(kv, TestRegionObserverInterface.B)) {
        foundB = true;
      }
      if (CellUtil.matchingFamily(kv, TestRegionObserverInterface.C)) {
        foundC = true;
      }
    }
    assertTrue(foundA);
    assertTrue(foundB);
    assertTrue(foundC);
  }
  ctPostGet.incrementAndGet();
}