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

@Override
public void map(WALKey key, WALEdit value,
  Context context)
throws IOException {
  try {
    // skip all other tables
    if (Bytes.equals(table, key.getTablename().getName())) {
      for (Cell cell : value.getCells()) {
        KeyValue kv = KeyValueUtil.ensureKeyValueTypeForMR(cell);
        if (WALEdit.isMetaEditFamily(kv.getFamily())) continue;
        context.write(new ImmutableBytesWritable(kv.getRow()), kv);
      }
    }
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
}

/**
 * @param row  The current table row key.
 * @param value  The columns.
 * @param context  The current context.
 * @throws IOException When something is broken with the data.
 */
@Override
public void map(ImmutableBytesWritable row, Result value,
  Context context)
throws IOException {
  try {
    if (LOG.isTraceEnabled()) {
      LOG.trace("Considering the row."
          + Bytes.toString(row.get(), row.getOffset(), row.getLength()));
    }
    if (filter == null || !filter.filterRowKey(row.get(), row.getOffset(), row.getLength())) {
      for (Cell kv : value.rawCells()) {
        kv = filterKv(filter, kv);
        // skip if we filtered it out
        if (kv == null) continue;
        // TODO get rid of ensureKeyValue
        context.write(row, KeyValueUtil.ensureKeyValueTypeForMR(convertKv(kv, cfRenameMap)));
      }
    }
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
}

/**
 * Checks whether the given scan rowkey range overlaps with the current storefile's
 *
 * @param scan the scan specification. Used to determine the rowkey range.
 * @return true if there is overlap, false otherwise
 */
public boolean passesKeyRangeFilter(Scan scan) {
  if (this.getFirstKey() == null || this.getLastKey() == null) {
    // the file is empty
    return false;
  }
  if (Bytes.equals(scan.getStartRow(), HConstants.EMPTY_START_ROW) && Bytes
      .equals(scan.getStopRow(), HConstants.EMPTY_END_ROW)) {
    return true;
  }
  KeyValue smallestScanKeyValue = scan.isReversed() ?
      KeyValueUtil.createFirstOnRow(scan.getStopRow()) :
      KeyValueUtil.createFirstOnRow(scan.getStartRow());
  KeyValue largestScanKeyValue = scan.isReversed() ?
      KeyValueUtil.createLastOnRow(scan.getStartRow()) :
      KeyValueUtil.createLastOnRow(scan.getStopRow());
  boolean nonOverLapping =
      (getComparator().compareFlatKey(this.getFirstKey(), largestScanKeyValue.getKey()) > 0
          && !Bytes.equals(scan.isReversed() ? scan.getStartRow() : scan.getStopRow(),
          HConstants.EMPTY_END_ROW))
          || getComparator().compareFlatKey(this.getLastKey(), smallestScanKeyValue.getKey())
          < 0;
  return !nonOverLapping;
}

@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 synchronized boolean seekToLastRow() {
  Cell first = cellSetAtCreation.isEmpty() ? null : cellSetAtCreation
      .last();
  Cell second = snapshotAtCreation.isEmpty() ? null
      : snapshotAtCreation.last();
  Cell higherCell = getHighest(first, second);
  if (higherCell == null) {
    return false;
  }
  Cell firstCellOnLastRow = KeyValueUtil.createFirstOnRow(higherCell.getRowArray(),
      higherCell.getRowOffset(), higherCell.getRowLength());
  if (seek(firstCellOnLastRow)) {
    return true;
  } else {
    return seekToPreviousRow(higherCell);
  }

}

/**
 * @param currentRow
 * @param offset
 * @param length
 * @return true when the joined heap may have data for the current row
 * @throws IOException
 */
private boolean joinedHeapMayHaveData(byte[] currentRow, int offset, short length)
    throws IOException {
  Cell nextJoinedKv = joinedHeap.peek();
  boolean matchCurrentRow =
      nextJoinedKv != null && CellUtil.matchingRow(nextJoinedKv, currentRow, offset, length);
  boolean matchAfterSeek = false;

  // If the next value in the joined heap does not match the current row,
  // try to seek to the
  // correct row
  if (!matchCurrentRow) {
    Cell firstOnCurrentRow = KeyValueUtil.createFirstOnRow(currentRow, offset, length);
    boolean seekSuccessful = this.joinedHeap.requestSeek(firstOnCurrentRow, true, true);
    matchAfterSeek = seekSuccessful && joinedHeap.peek() != null && CellUtil
        .matchingRow(joinedHeap.peek(), currentRow, offset, length);
  }

  return matchCurrentRow || matchAfterSeek;
}

@Override public synchronized boolean reseek(byte[] row) throws IOException {
  if (row == null) {
    throw new IllegalArgumentException("Row cannot be null.");
  }
  boolean result = false;
  startRegionOperation();
  KeyValue kv = KeyValueUtil.createFirstOnRow(row);
  try {
    // use request seek to make use of the lazy seek option. See HBASE-5520
    result = this.storeHeap.requestSeek(kv, true, true);
    if (this.joinedHeap != null) {
      result = this.joinedHeap.requestSeek(kv, true, true) || result;
    }
  } catch (FileNotFoundException e) {
    throw handleFileNotFound(e);
  } finally {
    closeRegionOperation();
  }
  return result;
}

@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;
}

private void assertScannerResults(KeyValueScanner scanner, KeyValue[] expected)
    throws IOException {
  scanner.seek(KeyValueUtil.createFirstOnRow(new byte[]{}));
  List<Cell> returned = Lists.newArrayList();

  while (true) {
    Cell next = scanner.next();
    if (next == null) break;
    returned.add(next);
  }

  assertTrue(
      "Got:\n" + Joiner.on("\n").join(returned) +
      "\nExpected:\n" + Joiner.on("\n").join(expected),
      Iterables.elementsEqual(Arrays.asList(expected), returned));
  assertNull(scanner.peek());
}

/**
 * Test for HBASE-8012
 */
public void testReseek() throws Exception {
  // write the file
  Path f = new Path(ROOT_DIR, getName());
  HFileContext meta = new HFileContextBuilder().withBlockSize(8 * 1024).build();
  // Make a store file and write data to it.
  StoreFile.Writer writer = new StoreFile.WriterBuilder(conf, cacheConf, this.fs)
          .withFilePath(f)
          .withFileContext(meta)
          .build();

  writeStoreFile(writer);
  writer.close();

  StoreFile.Reader reader = new StoreFile.Reader(fs, f, cacheConf, conf);

  // Now do reseek with empty KV to position to the beginning of the file

  KeyValue k = KeyValueUtil.createFirstOnRow(HConstants.EMPTY_BYTE_ARRAY);
  StoreFileScanner s = reader.getStoreFileScanner(false, false);
  s.reseek(k);

  assertNotNull("Intial reseek should position at the beginning of the file", s.peek());
}

/**
 * Test seeking while file is encoded.
 */
@Test
public void testSeekToBlockWithDecreasingCommonPrefix() throws IOException {
  List<KeyValue> sampleKv = new ArrayList<KeyValue>();
  KeyValue kv1 = new KeyValue(Bytes.toBytes("row10aaa"), Bytes.toBytes("f1"),
      Bytes.toBytes("q1"), Bytes.toBytes("val"));
  sampleKv.add(kv1);
  KeyValue kv2 = new KeyValue(Bytes.toBytes("row10aaa"), Bytes.toBytes("f1"),
      Bytes.toBytes("q2"), Bytes.toBytes("val"));
  sampleKv.add(kv2);
  KeyValue kv3 = new KeyValue(Bytes.toBytes("row10aaa"), Bytes.toBytes("f1"),
      Bytes.toBytes("q3"), Bytes.toBytes("val"));
  sampleKv.add(kv3);
  KeyValue kv4 = new KeyValue(Bytes.toBytes("row11baa"), Bytes.toBytes("f1"),
      Bytes.toBytes("q1"), Bytes.toBytes("val"));
  sampleKv.add(kv4);
  KeyValue toSeek = KeyValueUtil.createLastOnRow(kv3.getRowArray(), kv3.getRowOffset(),
      kv3.getRowLength(), null, 0, 0, null, 0, 0);
  seekToTheKey(kv3, sampleKv, toSeek);
}

protected int binarySearch(final Cell [] kvs,
                           final byte [] family,
                           final byte [] qualifier) {
  Cell searchTerm =
      KeyValueUtil.createFirstOnRow(CellUtil.cloneRow(kvs[0]),
          family, qualifier);

  // 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;
}

@Test
public void testScanBackwards() throws IOException {
  CellSearcher searcher = null;
  try {
    searcher = DecoderFactory.checkOut(block, true);
    searcher.positionAfterLastCell();
    int i = -1;
    while (searcher.previous()) {
      ++i;
      int oppositeIndex = rows.getInputs().size() - i - 1;
      KeyValue inputKv = rows.getInputs().get(oppositeIndex);
      KeyValue outputKv = KeyValueUtil.copyToNewKeyValue(searcher.current());
      Assert.assertEquals(inputKv, outputKv);
    }
    Assert.assertEquals(rows.getInputs().size(), i + 1);
  } finally {
    DecoderFactory.checkIn(searcher);
  }
}

@Test
public void testSeekWithPrefix() throws IOException {
  if (!(rows instanceof TestRowDataSearchWithPrefix)) {
    return;
  }
  CellSearcher searcher = null;
  try {
    searcher = DecoderFactory.checkOut(block, true);
    // seek with half bytes of second row key, should return second row
    KeyValue kv = rows.getInputs().get(1);
    KeyValue firstKVOnRow = KeyValueUtil.createFirstOnRow(Arrays.copyOfRange(
        kv.getRowArray(), kv.getRowOffset(),
        kv.getRowOffset() + kv.getRowLength() / 2));
    CellScannerPosition position = searcher.positionAtOrAfter(firstKVOnRow);
    Assert.assertEquals(CellScannerPosition.AFTER, position);
    Assert.assertEquals(kv, searcher.current());
  } finally {
    DecoderFactory.checkIn(searcher);
  }
}

/**
 * Exercise the nubCellsRemain variable by calling next+previous.  NubCellsRemain is basically
 * a special fan index.
 */
@Test
public void testReverseScannerWithJitter() {
  searcher.positionAfterLastCell();
  int counter = -1;
  while (true) {
    boolean foundCell = searcher.previous();
    if (!foundCell) {
      break;
    }
    ++counter;

    // a next+previous should cancel out
    if (!searcher.isAfterLast()) {
      searcher.advance();
      searcher.previous();
    }

    int oppositeIndex = rows.getInputs().size() - counter - 1;
    KeyValue inputKv = rows.getInputs().get(oppositeIndex);
    KeyValue outputKv = KeyValueUtil.copyToNewKeyValue(searcher.current());
    assertKeyAndValueEqual(inputKv, outputKv);
  }
  Assert.assertEquals(rows.getInputs().size(), counter + 1);
}

@Override
public int internalEncode(Cell cell, HFileBlockDefaultEncodingContext encodingContext,
    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 size = klength + vlength + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE;
  // Write the additional tag into the stream
  if (encodingContext.getHFileContext().isIncludesTags()) {
    int tagsLength = cell.getTagsLength();
    out.writeShort(tagsLength);
    if (tagsLength > 0) {
      out.write(cell.getTagsArray(), cell.getTagsOffset(), tagsLength);
    }
    size += tagsLength + KeyValue.TAGS_LENGTH_SIZE;
  }
  if (encodingContext.getHFileContext().isIncludesMvcc()) {
    WritableUtils.writeVLong(out, cell.getSequenceId());
    size += WritableUtils.getVIntSize(cell.getSequenceId());
  }
  return size;
}

public static long calculateMutationSize(final Mutation mutation) {
  long size = 0;
  for (Map.Entry<byte[], List<Cell>> entry : mutation.getFamilyCellMap().entrySet()) {
    for (Cell cell : entry.getValue()) {
      size += KeyValueUtil.length(cell);
    }
  }
  return size;
}

public static long calculateResultSize(final Result result) {
  long size = 0;
  for (Cell cell : result.rawCells()) {
    size += KeyValueUtil.length(cell);
  }
  return size;
}

public static long calculateResultSize(final List<Result> results) {
  long size = 0;
  for (Result result : results) {
    for (Cell cell : result.rawCells()) {
      size += KeyValueUtil.length(cell);
    }
  }
  return size;
}

/**
 * Prepare an ordered pair of row and qualifier to be compared using
 * KeyValue.KeyComparator. This is only used for row-column Bloom
 * filters.
 */
@Override
public byte[] createBloomKey(byte[] row, int roffset, int rlength,
    byte[] qualifier, int qoffset, int qlength) {
  if (qualifier == null)
    qualifier = DUMMY;

  // Make sure this does not specify a timestamp so that the default maximum
  // (most recent) timestamp is used.
  KeyValue kv = KeyValueUtil.createFirstOnRow(row, roffset, rlength, DUMMY, 0, 0,
      qualifier, qoffset, qlength);
  return kv.getKey();
}

@Override
public void write(Cell cell) throws IOException {
  // We first write the KeyValue infrastructure as VInts.
  StreamUtils.writeRawVInt32(out, KeyValueUtil.keyLength(cell));
  StreamUtils.writeRawVInt32(out, cell.getValueLength());
  // To support tags
  int tagsLength = cell.getTagsLength();
  StreamUtils.writeRawVInt32(out, tagsLength);

  // Write row, qualifier, and family; use dictionary
  // compression as they're likely to have duplicates.
  write(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(), compression.rowDict);
  write(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
      compression.familyDict);
  write(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
      compression.qualifierDict);

  // Write timestamp, type and value as uncompressed.
  StreamUtils.writeLong(out, cell.getTimestamp());
  out.write(cell.getTypeByte());
  out.write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  if (tagsLength > 0) {
    if (compression.tagCompressionContext != null) {
      // Write tags using Dictionary compression
      compression.tagCompressionContext.compressTags(out, cell.getTagsArray(),
          cell.getTagsOffset(), tagsLength);
    } else {
      // Tag compression is disabled within the WAL compression. Just write the tags bytes as
      // it is.
      out.write(cell.getTagsArray(), cell.getTagsOffset(), tagsLength);
    }
  }
}

private void winterTestingStoreFile(StoreFile sf) throws IOException {
  StoreFileScanner compactedFileScanner = sf.getReader().getStoreFileScanner(false, false);
  KeyValue startKey =
      KeyValueUtil.createFirstOnRow(HConstants.EMPTY_START_ROW, HConstants.LATEST_TIMESTAMP);
  compactedFileScanner.seek(startKey);
  KeyValue kv;
  int n = 0;
  while ((kv = (KeyValue) compactedFileScanner.next()) != null) {
    LOG.info("LCDBG, show kv: " + Bytes.toInt(kv.getRow()));
    ++n;
  }
  LOG.info("LCDBG, reader has: " + n + " in " + sf.getPath());
  compactedFileScanner.close();
}

public Cell getKeyForNextColumn(Cell kv) {
  ColumnCount nextColumn = columns.getColumnHint();
  if (nextColumn == null) {
    return KeyValueUtil.createLastOnRow(
        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength(),
        kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength(),
        kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength());
  } else {
    return KeyValueUtil.createFirstOnRow(
        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength(),
        kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength(),
        nextColumn.getBuffer(), nextColumn.getOffset(), nextColumn.getLength());
  }
}

@Override public boolean seekToLastRow() throws IOException {
  byte[] lastRow = reader.getLastRowKey();
  if (lastRow == null) {
    return false;
  }
  KeyValue seekKey = KeyValueUtil.createFirstOnRow(lastRow);
  if (seek(seekKey)) {
    return true;
  } else {
    return seekToPreviousRow(seekKey);
  }
}

private void addDelete(final Cell kv) {
  NavigableSet<KeyValue> rowdeletes = this.deletes.get(kv);
  if (rowdeletes == null) {
    rowdeletes = new TreeSet<KeyValue>(this.kvcomparator);
    this.deletes.put(KeyValueUtil.ensureKeyValue(kv), rowdeletes);
  }
  rowdeletes.add(KeyValueUtil.ensureKeyValue(kv));
}

/**
 * Do right thing with passed key, add to deletes or add to candidates.
 * @param kv
 * @return True if we added a candidate
 */
boolean handle(final Cell kv) {
  if (KeyValueUtil.ensureKeyValue(kv).isDelete()) {
    handleDeletes(kv);
    return false;
  }
  return addCandidate(kv);
}

/**
 * Separately get the KeyValue before the specified key from kvset and
 * snapshotset, and use the row of higher one as the previous row of
 * specified key, then seek to the first KeyValue of previous row
 */
@Override
public synchronized boolean seekToPreviousRow(Cell key) {
  Cell firstKeyOnRow = KeyValueUtil.createFirstOnRow(key.getRowArray(), key.getRowOffset(),
      key.getRowLength());
  SortedSet<Cell> cellHead = cellSetAtCreation.headSet(firstKeyOnRow);
  Cell cellSetBeforeRow = cellHead.isEmpty() ? null : cellHead.last();
  SortedSet<Cell> snapshotHead = snapshotAtCreation
      .headSet(firstKeyOnRow);
  Cell snapshotBeforeRow = snapshotHead.isEmpty() ? null : snapshotHead
      .last();
  Cell lastCellBeforeRow = getHighest(cellSetBeforeRow, snapshotBeforeRow);
  if (lastCellBeforeRow == null) {
    theNext = null;
    return false;
  }
  Cell firstKeyOnPreviousRow = KeyValueUtil.createFirstOnRow(lastCellBeforeRow.getRowArray(),
      lastCellBeforeRow.getRowOffset(), lastCellBeforeRow.getRowLength());
  this.stopSkippingCellsIfNextRow = true;
  seek(firstKeyOnPreviousRow);
  this.stopSkippingCellsIfNextRow = false;
  if (peek() == null
      || comparator.compareRows(peek(), firstKeyOnPreviousRow) > 0) {
    return seekToPreviousRow(lastCellBeforeRow);
  }
  return true;
}

public void collect(Cell cell) {
  valLen.update(cell.getValueLength());
  if (prevCell != null &&
      KeyValue.COMPARATOR.compareRows(prevCell, cell) != 0) {
    // new row
    collectRow();
  }
  curRowBytes += KeyValueUtil.length(cell);
  curRowKeyLength = KeyValueUtil.keyLength(cell);
  curRowCols++;
  prevCell = cell;
}

@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);
  }
}

@Override
public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put m, WALEdit edit,
    Durability durability) throws IOException {
  byte[] attribute = m.getAttribute(NON_VISIBILITY);
  byte[] cf = null;
  List<Cell> updatedCells = new ArrayList<Cell>();
  if (attribute != null) {
    for (List<? extends Cell> edits : m.getFamilyCellMap().values()) {
      for (Cell cell : edits) {
        KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
        if (cf == null) {
          cf = kv.getFamily();
        }
        Tag tag = new Tag((byte) NON_VIS_TAG_TYPE, attribute);
        List<Tag> tagList = new ArrayList<Tag>();
        tagList.add(tag);
        tagList.addAll(kv.getTags());
        byte[] fromList = Tag.fromList(tagList);
        TagRewriteCell newcell = new TagRewriteCell(kv, fromList);
        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(newcell);
      }
    }
    m.getFamilyCellMap().remove(cf);
    // Update the family map
    m.getFamilyCellMap().put(cf, updatedCells);
  }
}

@Test
public void testReseek() throws Exception {
  // write the file
  Path f = new Path(ROOT_DIR, "testReseek");
  HFileContext meta = new HFileContextBuilder().withBlockSize(8 * 1024).withIncludesTags(true)
      .withCompressTags(true).withDataBlockEncoding(DataBlockEncoding.PREFIX).build();
  // Make a store file and write data to it.
  StoreFile.Writer writer = new StoreFile.WriterBuilder(conf, cacheConf, fs).withFilePath(f)
      .withFileContext(meta).build();

  writeStoreFile(writer);
  writer.close();

  StoreFile.Reader reader = new StoreFile.Reader(fs, f, cacheConf, conf);
  StoreFileScanner s = reader.getStoreFileScanner(false, false);
  try {
    // Now do reseek with empty KV to position to the beginning of the file
    KeyValue k = KeyValueUtil.createFirstOnRow(Bytes.toBytes("k2"));
    s.reseek(k);
    Cell kv = s.next();
    kv = s.next();
    kv = s.next();
    byte[] key5 = Bytes.toBytes("k5");
    assertTrue(Bytes.equals(key5, 0, key5.length, kv.getRowArray(), kv.getRowOffset(),
        kv.getRowLength()));
    List<Tag> tags = KeyValueUtil.ensureKeyValue(kv).getTags();
    assertEquals(1, tags.size());
    assertEquals("tag3", Bytes.toString(tags.get(0).getValue()));
  } finally {
    s.close();
  }
}

private void verifyScanAcrossSnapshot2(KeyValue kv1, KeyValue kv2)
    throws IOException {
  List<KeyValueScanner> memstorescanners = this.memstore.getScanners(mvcc.getReadPoint());
  assertEquals(1, memstorescanners.size());
  final KeyValueScanner scanner = memstorescanners.get(0);
  scanner.seek(KeyValueUtil.createFirstOnRow(HConstants.EMPTY_START_ROW));
  assertEquals(kv1, scanner.next());
  assertEquals(kv2, scanner.next());
  assertNull(scanner.next());
}

static void doScan(MemStore ms, int iteration) throws IOException {
  long nanos = System.nanoTime();
  KeyValueScanner s = ms.getScanners(0).get(0);
  s.seek(KeyValueUtil.createFirstOnRow(new byte[]{}));

  System.out.println(iteration + " create/seek took: " + (System.nanoTime() - nanos)/1000);
  int cnt=0;
  while(s.next() != null) ++cnt;

  System.out.println(iteration + " took usec: " + (System.nanoTime() - nanos) / 1000 + " for: "
      + cnt);

}

private void updateMutationAddingTags(final Mutation m) {
  byte[] attribute = m.getAttribute("visibility");
  byte[] cf = null;
  List<Cell> updatedCells = new ArrayList<Cell>();
  if (attribute != null) {
    for (List<? extends Cell> edits : m.getFamilyCellMap().values()) {
      for (Cell cell : edits) {
        KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
        if (cf == null) {
          cf = kv.getFamily();
        }
        Tag tag = new Tag((byte) 1, 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);
      }
    }
    m.getFamilyCellMap().remove(cf);
    // Update the family map
    m.getFamilyCellMap().put(cf, updatedCells);
  }
}

@Override
public void visitLogEntryBeforeWrite(HTableDescriptor htd, WALKey logKey, WALEdit logEdit) {
  for (Cell cell : logEdit.getCells()) {
    KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
    for (Map.Entry entry : kv.toStringMap().entrySet()) {
      if (entry.getValue().equals(Bytes.toString(WALEdit.BULK_LOAD))) {
        found = true;
      }
    }
  }
}

public void testKeyValueScanFixture() throws IOException {
  KeyValue kvs[] = new KeyValue[]{
      KeyValueTestUtil.create("RowA", "family", "qf1",
          1, KeyValue.Type.Put, "value-1"),
      KeyValueTestUtil.create("RowA", "family", "qf2",
          1, KeyValue.Type.Put, "value-2"),
      KeyValueTestUtil.create("RowB", "family", "qf1",
          10, KeyValue.Type.Put, "value-10")
  };
  KeyValueScanner scan = new KeyValueScanFixture(
      KeyValue.COMPARATOR, kvs);

  KeyValue kv = KeyValueUtil.createFirstOnRow(Bytes.toBytes("RowA"));
  // should seek to this:
  assertTrue(scan.seek(kv));
  Cell res = scan.peek();
  assertEquals(kvs[0], res);

  kv = KeyValueUtil.createFirstOnRow(Bytes.toBytes("RowB"));
  assertTrue(scan.seek(kv));
  res = scan.peek();
  assertEquals(kvs[2], res);

  // ensure we pull things out properly:
  kv = KeyValueUtil.createFirstOnRow(Bytes.toBytes("RowA"));
  assertTrue(scan.seek(kv));
  assertEquals(kvs[0], scan.peek());
  assertEquals(kvs[0], scan.next());
  assertEquals(kvs[1], scan.peek());
  assertEquals(kvs[1], scan.next());
  assertEquals(kvs[2], scan.peek());
  assertEquals(kvs[2], scan.next());
  assertEquals(null, scan.peek());
  assertEquals(null, scan.next());
}

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);
}

private static void assertNext(boolean reverse, byte[] fuzzyRow, byte[] mask, byte[] current,
    byte[] expected) {
  KeyValue kv = KeyValueUtil.createFirstOnRow(current);
  byte[] nextForFuzzyRule = FuzzyRowFilter.getNextForFuzzyRule(reverse, kv.getRowArray(),
      kv.getRowOffset(), kv.getRowLength(), fuzzyRow, mask);
  Assert.assertEquals(Bytes.toStringBinary(expected), Bytes.toStringBinary(nextForFuzzyRule));
}

/**
 * Tests the behavior of transform() in a hierarchical filter.
 *
 * transform() only applies after a filterKeyValue() whose return-code includes the KeyValue.
 * Lazy evaluation of AND
 */
@Test
public void testTransformMPO() throws Exception {
  // Apply the following filter:
  //     (family=fam AND qualifier=qual1 AND KeyOnlyFilter)
  //  OR (family=fam AND qualifier=qual2)
  final FilterList flist = new FilterList(Operator.MUST_PASS_ONE, Lists.<Filter>newArrayList(
      new FilterList(Operator.MUST_PASS_ALL, Lists.<Filter>newArrayList(
          new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("fam"))),
          new QualifierFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("qual1"))),
          new KeyOnlyFilter())),
      new FilterList(Operator.MUST_PASS_ALL, Lists.<Filter>newArrayList(
          new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("fam"))),
          new QualifierFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("qual2")))))));

  final KeyValue kvQual1 = new KeyValue(
      Bytes.toBytes("row"), Bytes.toBytes("fam"), Bytes.toBytes("qual1"), Bytes.toBytes("value"));
  final KeyValue kvQual2 = new KeyValue(
      Bytes.toBytes("row"), Bytes.toBytes("fam"), Bytes.toBytes("qual2"), Bytes.toBytes("value"));
  final KeyValue kvQual3 = new KeyValue(
      Bytes.toBytes("row"), Bytes.toBytes("fam"), Bytes.toBytes("qual3"), Bytes.toBytes("value"));

  // Value for fam:qual1 should be stripped:
  assertEquals(Filter.ReturnCode.INCLUDE, flist.filterKeyValue(kvQual1));
  final KeyValue transformedQual1 = KeyValueUtil.ensureKeyValue(flist.transform(kvQual1));
  assertEquals(0, transformedQual1.getValue().length);

  // Value for fam:qual2 should not be stripped:
  assertEquals(Filter.ReturnCode.INCLUDE, flist.filterKeyValue(kvQual2));
  final KeyValue transformedQual2 = KeyValueUtil.ensureKeyValue(flist.transform(kvQual2));
  assertEquals("value", Bytes.toString(transformedQual2.getValue()));

  // Other keys should be skipped:
  assertEquals(Filter.ReturnCode.SKIP, flist.filterKeyValue(kvQual3));
}

@Override
public void write(Cell c) throws IOException {
  KeyValue kv = KeyValueUtil.ensureKeyValue(c);
  expectState(State.WRITING);
  this.dataBlockEncoder.encode(kv, dataBlockEncodingCtx, this.userDataStream);
  this.unencodedDataSizeWritten += kv.getLength();
  if (dataBlockEncodingCtx.getHFileContext().isIncludesMvcc()) {
    this.unencodedDataSizeWritten += WritableUtils.getVIntSize(kv.getMvccVersion());
  }
}