public HFileSortedIterator(HFileScanner scan, SerializedComparator comparator, 
    byte[] from, boolean fromInclusive, 
    byte[] to, boolean toInclusive) throws IOException {
  this.scan = scan;
  this.comparator = comparator;
  this.from = from;
  this.fromInclusive = fromInclusive;
  this.to = to;
  this.toInclusive = toInclusive;
  
  assert from == null 
      || to == null 
      || comparator.compare(from, 0, from.length, to, 0, to.length) <= 0;
  
  start = sopConfig.getStatistics().getScan().begin();
  foundNext = evalFrom();
}
 

public HFileSortedIterator(HFileScanner scan, byte[] from, boolean fromInclusive, byte[] to, 
    boolean toInclusive) throws IOException {
  this.scan = scan;
  this.from = from;
  this.fromInclusive = fromInclusive;
  this.to = to;
  this.toInclusive = toInclusive;

  scanStat = (stats == null) ? new SortedOplogStatistics("", "").new ScanOperation(
      0, 0, 0, 0, 0, 0, 0) : stats.getScan();
  scanStart = scanStat.begin();

  if (scan == null) {
    return;
  }

  assert from == null || to == null
      || scan.getReader().getComparator().compare(from, to) <= 0;

  initIterator();
}
 

private int readAndCheckbytes(HFileScanner scanner, int start, int n)
    throws IOException {
  String value = "value";
  int i = start;
  for (; i < (start + n); i++) {
    ByteBuffer key = scanner.getKey();
    ByteBuffer val = scanner.getValue();
    String keyStr = String.format(LOCAL_FORMATTER, Integer.valueOf(i));
    String valStr = value + keyStr;
    KeyValue kv = new KeyValue(Bytes.toBytes(keyStr), Bytes.toBytes("family"),
        Bytes.toBytes("qual"), Bytes.toBytes(valStr));
    byte[] keyBytes = new KeyValue.KeyOnlyKeyValue(Bytes.toBytes(key), 0,
        Bytes.toBytes(key).length).getKey();
    assertArrayEquals(kv.getKey(), keyBytes,
        "bytes for keys do not match " + keyStr + " " + Bytes.toString(Bytes.toBytes(key)));
    byte[] valBytes = Bytes.toBytes(val);
    assertArrayEquals(Bytes.toBytes(valStr), valBytes,
        "bytes for vals do not match " + valStr + " " + Bytes.toString(valBytes));
    if (!scanner.next()) {
      break;
    }
  }
  assertEquals(i, start + n - 1);
  return (start + n);
}
 

/**
 * Issues an HFile lookup on the underlying HFile.Reader. This is protected
 * for testing.
 */
protected Pair<ByteBuffer, Pair<ByteBuffer, Throwable>> getValueFromHFile(ByteBuffer key) {
  try {
    HFileScanner scanner = reader.getScanner(true, true, false);
    KeyValue kv = buildKeyValueForLookup(
        BytesUtil.readBytesFromByteBufferWithoutConsume(key));
    int code = scanner.seekTo(kv.getKey());
    ByteBuffer value = null;
    if (code == 0) {
      value = ByteBuffer.wrap(scanner.getKeyValue().getValue());
      if (this.sizeStatsKey != null) {
        Stats.addMetric(this.sizeStatsKey, value.remaining());
      }
      Stats.addMetric("value-size", value.remaining());
    } else {
      Stats.incr("not-found");
      if (this.notFoundStatsKey != null) {
        Stats.incr(this.notFoundStatsKey);
      }
    }
    return new ImmutablePair(key, new ImmutablePair(value, null));
  } catch (Throwable t) {
    return new ImmutablePair(key, new ImmutablePair(null, t));
  }
}
 

public HFileSortedIterator(HFileScanner scan, SerializedComparator comparator, 
    byte[] from, boolean fromInclusive, 
    byte[] to, boolean toInclusive) throws IOException {
  this.scan = scan;
  this.comparator = comparator;
  this.from = from;
  this.fromInclusive = fromInclusive;
  this.to = to;
  this.toInclusive = toInclusive;
  
  assert from == null 
      || to == null 
      || comparator.compare(from, 0, from.length, to, 0, to.length) <= 0;
  
  start = sopConfig.getStatistics().getScan().begin();
  foundNext = evalFrom();
}
 

public HFileSortedIterator(HFileScanner scan, byte[] from, boolean fromInclusive, byte[] to, 
    boolean toInclusive) throws IOException {
  this.scan = scan;
  this.from = from;
  this.fromInclusive = fromInclusive;
  this.to = to;
  this.toInclusive = toInclusive;

  scanStat = (stats == null) ? new SortedOplogStatistics("", "").new ScanOperation(
      0, 0, 0, 0, 0, 0, 0) : stats.getScan();
  scanStart = scanStat.begin();

  if (scan == null) {
    return;
  }

  assert from == null || to == null
      || scan.getReader().getComparator().compare(from, to) <= 0;

  initIterator();
}
 

/**
 *
 * @param s
 * @param k
 * @return false if not found or if k is after the end.
 * @throws IOException
 */
public static boolean seekAtOrAfter(HFileScanner s, Cell k)
throws IOException {
  int result = s.seekTo(k);
  if(result < 0) {
    if (result == HConstants.INDEX_KEY_MAGIC) {
      // using faked key
      return true;
    }
    // Passed KV is smaller than first KV in file, work from start of file
    return s.seekTo();
  } else if(result > 0) {
    // Passed KV is larger than current KV in file, if there is a next
    // it is the "after", if not then this scanner is done.
    return s.next();
  }
  // Seeked to the exact key
  return true;
}
 

static boolean reseekAtOrAfter(HFileScanner s, Cell k)
throws IOException {
  //This function is similar to seekAtOrAfter function
  int result = s.reseekTo(k);
  if (result <= 0) {
    if (result == HConstants.INDEX_KEY_MAGIC) {
      // using faked key
      return true;
    }
    // If up to now scanner is not seeked yet, this means passed KV is smaller
    // than first KV in file, and it is the first time we seek on this file.
    // So we also need to work from the start of file.
    if (!s.isSeeked()) {
      return  s.seekTo();
    }
    return true;
  }
  // passed KV is larger than current KV in file, if there is a next
  // it is after, if not then this scanner is done.
  return s.next();
}
 

@Override
public Optional<Cell> getLastKey() {
  if (top) {
    return super.getLastKey();
  }
  // Get a scanner that caches the block and that uses pread.
  HFileScanner scanner = getScanner(true, true);
  try {
    if (scanner.seekBefore(this.splitCell)) {
      return Optional.ofNullable(scanner.getKey());
    }
  } catch (IOException e) {
    LOG.warn("Failed seekBefore " + Bytes.toStringBinary(this.splitkey), e);
  } finally {
    if (scanner != null) {
      scanner.close();
    }
  }
  return Optional.empty();
}
 

@Override
public Optional<Cell> getFirstKey() {
  if (!firstKeySeeked) {
    HFileScanner scanner = getScanner(true, true, false);
    try {
      if (scanner.seekTo()) {
        this.firstKey = Optional.ofNullable(scanner.getKey());
      }
      firstKeySeeked = true;
    } catch (IOException e) {
      LOG.warn("Failed seekTo first KV in the file", e);
    } finally {
      if(scanner != null) {
        scanner.close();
      }
    }
  }
  return this.firstKey;
}
 

@Override
void doRow(int i) throws Exception {
  HFileScanner scanner = this.reader.getScanner(false, false);
  byte [] b = getRandomRow();
  // System.out.println("Random row: " + new String(b));
  Cell c = createCell(b);
  if (scanner.seekTo(c) != 0) {
    LOG.info("Nonexistent row: " + new String(b));
    return;
  }
  // TODO: HFileScanner doesn't do Cells yet. Temporary fix.
  c = scanner.getCell();
  // System.out.println("Found row: " +
  //  new String(c.getRowArray(), c.getRowOffset(), c.getRowLength()));
  PerformanceEvaluationCommons.assertKey(b, c);
  for (int ii = 0; ii < 30; ii++) {
    if (!scanner.next()) {
      LOG.info("NOTHING FOLLOWS");
      return;
    }
    c = scanner.getCell();
    PerformanceEvaluationCommons.assertValueSize(c.getValueLength(), ROW_LENGTH);
  }
}
 

private void verifyCounts(int countRow1, int countRow2) throws Exception {
  int count1 = 0;
  int count2 = 0;
  for (HStoreFile f : r.getStore(COLUMN_FAMILY_TEXT).getStorefiles()) {
    HFileScanner scanner = f.getReader().getScanner(false, false);
    scanner.seekTo();
    do {
      byte[] row = CellUtil.cloneRow(scanner.getCell());
      if (Bytes.equals(row, STARTROW)) {
        count1++;
      } else if (Bytes.equals(row, secondRowBytes)) {
        count2++;
      }
    } while (scanner.next());
  }
  assertEquals(countRow1, count1);
  assertEquals(countRow2, count2);
}
 

/**
 * Creates an inner HFileScanner object for a given HFile path
 */
public static HFileScanner createScanner(FileSystem fs, Path path) throws IOException {
    Configuration config = fs.getConf();
    HFile.Reader reader = HFile.createReader(fs, path, getCacheConfig(config), config);

    HFileScanner scanner = reader.getScanner(false, false);
    scanner.seekTo();

    return scanner;
}
 

public static void doSmokeTest(FileSystem fs, Path path, String codec)
throws Exception {
  Configuration conf = HBaseConfiguration.create();
  HFileContext context = new HFileContextBuilder()
                         .withCompression(HFileWriterImpl.compressionByName(codec)).build();
  HFile.Writer writer = HFile.getWriterFactoryNoCache(conf)
      .withPath(fs, path)
      .withFileContext(context)
      .create();
  // Write any-old Cell...
  final byte [] rowKey = Bytes.toBytes("compressiontestkey");
  Cell c = ExtendedCellBuilderFactory.create(CellBuilderType.DEEP_COPY)
    .setRow(rowKey)
    .setFamily(HConstants.EMPTY_BYTE_ARRAY)
    .setQualifier(HConstants.EMPTY_BYTE_ARRAY)
    .setTimestamp(HConstants.LATEST_TIMESTAMP)
    .setType(KeyValue.Type.Maximum.getCode())
    .setValue(Bytes.toBytes("compressiontestval"))
    .build();
  writer.append(c);
  writer.appendFileInfo(Bytes.toBytes("compressioninfokey"), Bytes.toBytes("compressioninfoval"));
  writer.close();
  Cell cc = null;
  HFile.Reader reader = HFile.createReader(fs, path, CacheConfig.DISABLED, true, conf);
  try {
    HFileScanner scanner = reader.getScanner(false, true);
    scanner.seekTo(); // position to the start of file
    // Scanner does not do Cells yet. Do below for now till fixed.
    cc = scanner.getCell();
    if (CellComparator.getInstance().compareRows(c, cc) != 0) {
      throw new Exception("Read back incorrect result: " + c.toString() + " vs " + cc.toString());
    }
  } finally {
    reader.close();
  }
}
 

/**
 * Implements a {@link KeyValueScanner} on top of the specified {@link HFileScanner}
 * @param useMVCC If true, scanner will filter out updates with MVCC larger than {@code readPt}.
 * @param readPt MVCC value to use to filter out the updates newer than this scanner.
 * @param hasMVCC Set to true if underlying store file reader has MVCC info.
 * @param scannerOrder Order of the scanner relative to other scanners. See
 *          {@link KeyValueScanner#getScannerOrder()}.
 * @param canOptimizeForNonNullColumn {@code true} if we can make sure there is no null column,
 *          otherwise {@code false}. This is a hint for optimization.
 */
public StoreFileScanner(StoreFileReader reader, HFileScanner hfs, boolean useMVCC,
    boolean hasMVCC, long readPt, long scannerOrder, boolean canOptimizeForNonNullColumn) {
  this.readPt = readPt;
  this.reader = reader;
  this.hfs = hfs;
  this.enforceMVCC = useMVCC;
  this.hasMVCCInfo = hasMVCC;
  this.scannerOrder = scannerOrder;
  this.canOptimizeForNonNullColumn = canOptimizeForNonNullColumn;
  this.reader.incrementRefCount();
}
 

private int verifyHFile(Path p) throws IOException {
  Configuration conf = util.getConfiguration();
  HFile.Reader reader =
    HFile.createReader(p.getFileSystem(conf), p, new CacheConfig(conf), true, conf);
  HFileScanner scanner = reader.getScanner(false, false);
  scanner.seekTo();
  int count = 0;
  do {
    count++;
  } while (scanner.next());
  assertTrue(count > 0);
  reader.close();
  return count;
}
 

@Override
void doRow(int i) throws Exception {
  HFileScanner scanner = this.reader.getScanner(false, true);
  byte [] b = getRandomRow();
  if (scanner.seekTo(createCell(b)) < 0) {
    LOG.info("Not able to seekTo " + new String(b));
    return;
  }
  // TODO: Fix scanner so it does Cells
  Cell c = scanner.getCell();
  PerformanceEvaluationCommons.assertKey(b, c);
  PerformanceEvaluationCommons.assertValueSize(c.getValueLength(), ROW_LENGTH);
}
 

@Override
void doRow(int i) throws Exception {
  HFileScanner scanner = this.reader.getScanner(false, true);
  byte[] gaussianRandomRowBytes = getGaussianRandomRowBytes();
  scanner.seekTo(createCell(gaussianRandomRowBytes));
  for (int ii = 0; ii < 30; ii++) {
    if (!scanner.next()) {
      LOG.info("NOTHING FOLLOWS");
      return;
    }
    // TODO: Fix. Make scanner do Cells.
    scanner.getCell();
  }
}
 

@Test
public void testStoreFileReference() throws Exception {
  final RegionInfo hri =
    RegionInfoBuilder.newBuilder(TableName.valueOf("testStoreFileReference")).build();
  HRegionFileSystem regionFs = HRegionFileSystem.createRegionOnFileSystem(conf, fs,
    new Path(testDir, hri.getTable().getNameAsString()), hri);
  HFileContext meta = new HFileContextBuilder().withBlockSize(8 * 1024).build();

  // Make a store file and write data to it.
  StoreFileWriter writer = new StoreFileWriter.Builder(conf, cacheConf, this.fs)
    .withFilePath(regionFs.createTempName()).withFileContext(meta).build();
  writeStoreFile(writer);
  Path hsfPath = regionFs.commitStoreFile(TEST_FAMILY, writer.getPath());
  writer.close();

  HStoreFile file = new HStoreFile(this.fs, hsfPath, conf, cacheConf, BloomType.NONE, true);
  file.initReader();
  StoreFileReader r = file.getReader();
  assertNotNull(r);
  StoreFileScanner scanner =
    new StoreFileScanner(r, mock(HFileScanner.class), false, false, 0, 0, false);

  // Verify after instantiating scanner refCount is increased
  assertTrue("Verify file is being referenced", file.isReferencedInReads());
  scanner.close();
  // Verify after closing scanner refCount is decreased
  assertFalse("Verify file is not being referenced", file.isReferencedInReads());
}
 

@Test
public void testEmptyStoreFileRestrictKeyRanges() throws Exception {
  StoreFileReader reader = mock(StoreFileReader.class);
  HStore store = mock(HStore.class);
  byte[] cf = Bytes.toBytes("ty");
  ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder.of(cf);
  when(store.getColumnFamilyDescriptor()).thenReturn(cfd);
  try (StoreFileScanner scanner =
    new StoreFileScanner(reader, mock(HFileScanner.class), false, false, 0, 0, true)) {
    Scan scan = new Scan();
    scan.setColumnFamilyTimeRange(cf, 0, 1);
    assertFalse(scanner.shouldUseScanner(scan, store, 0));
  }
}
 

@Test
public void testHFileLink() throws IOException {
  final RegionInfo hri =
    RegionInfoBuilder.newBuilder(TableName.valueOf("testHFileLinkTb")).build();
  // force temp data in hbase/target/test-data instead of /tmp/hbase-xxxx/
  Configuration testConf = new Configuration(this.conf);
  CommonFSUtils.setRootDir(testConf, testDir);
  HRegionFileSystem regionFs = HRegionFileSystem.createRegionOnFileSystem(testConf, fs,
    CommonFSUtils.getTableDir(testDir, hri.getTable()), hri);
  HFileContext meta = new HFileContextBuilder().withBlockSize(8 * 1024).build();

  // Make a store file and write data to it.
  StoreFileWriter writer = new StoreFileWriter.Builder(conf, cacheConf, this.fs)
    .withFilePath(regionFs.createTempName()).withFileContext(meta).build();
  writeStoreFile(writer);

  Path storeFilePath = regionFs.commitStoreFile(TEST_FAMILY, writer.getPath());
  Path dstPath = new Path(regionFs.getTableDir(), new Path("test-region", TEST_FAMILY));
  HFileLink.create(testConf, this.fs, dstPath, hri, storeFilePath.getName());
  Path linkFilePath =
    new Path(dstPath, HFileLink.createHFileLinkName(hri, storeFilePath.getName()));

  // Try to open store file from link
  StoreFileInfo storeFileInfo = new StoreFileInfo(testConf, this.fs, linkFilePath, true);
  HStoreFile hsf = new HStoreFile(storeFileInfo, BloomType.NONE, cacheConf);
  assertTrue(storeFileInfo.isLink());
  hsf.initReader();

  // Now confirm that I can read from the link
  int count = 1;
  HFileScanner s = hsf.getReader().getScanner(false, false);
  s.seekTo();
  while (s.next()) {
    count++;
  }
  assertEquals((LAST_CHAR - FIRST_CHAR + 1) * (LAST_CHAR - FIRST_CHAR + 1), count);
}
 

private void readStoreFile(Path path) throws IOException {
  CacheConfig cacheConf = store.getCacheConfig();
  BlockCache cache = cacheConf.getBlockCache().get();
  HStoreFile sf = new HStoreFile(fs, path, conf, cacheConf, BloomType.ROWCOL, true);
  sf.initReader();
  HFile.Reader reader = sf.getReader().getHFileReader();
  try {
    // Open a scanner with (on read) caching disabled
    HFileScanner scanner = reader.getScanner(false, false);
    assertTrue(testDescription, scanner.seekTo());
    // Cribbed from io.hfile.TestCacheOnWrite
    long offset = 0;
    while (offset < reader.getTrailer().getLoadOnOpenDataOffset()) {
      // Flags: don't cache the block, use pread, this is not a compaction.
      // Also, pass null for expected block type to avoid checking it.
      HFileBlock block = reader.readBlock(offset, -1, false, true,
        false, true, null, DataBlockEncoding.NONE);
      BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(),
        offset);
      boolean isCached = cache.getBlock(blockCacheKey, true, false, true) != null;
      boolean shouldBeCached = cowType.shouldBeCached(block.getBlockType());
      final BlockType blockType = block.getBlockType();

      if (shouldBeCached != isCached &&
          (cowType.blockType1.equals(blockType) || cowType.blockType2.equals(blockType))) {
        throw new AssertionError(
          "shouldBeCached: " + shouldBeCached+ "\n" +
          "isCached: " + isCached + "\n" +
          "Test description: " + testDescription + "\n" +
          "block: " + block + "\n" +
          "blockCacheKey: " + blockCacheKey);
      }
      offset += block.getOnDiskSizeWithHeader();
    }
  } finally {
    reader.close();
  }
}
 

private Cell doTestOfSeekBefore(Path p, FileSystem fs, Reference bottom, Cell seekBefore,
    CacheConfig cacheConfig) throws IOException {
  ReaderContext context = new ReaderContextBuilder().withFileSystemAndPath(fs, p).build();
  HFileInfo fileInfo = new HFileInfo(context, TEST_UTIL.getConfiguration());
  final HalfStoreFileReader halfreader = new HalfStoreFileReader(context, fileInfo, cacheConfig,
      bottom, new AtomicInteger(0), TEST_UTIL.getConfiguration());
  fileInfo.initMetaAndIndex(halfreader.getHFileReader());
  halfreader.loadFileInfo();
  final HFileScanner scanner = halfreader.getScanner(false, false);
  scanner.seekBefore(seekBefore);
  return scanner.getCell();
}
 

/**
 * Method returns the total KVs in given hfile
 * @param fs File System
 * @param p HFile path
 * @return KV count in the given hfile
 * @throws IOException
 */
private static int getKVCountFromHfile(FileSystem fs, Path p) throws IOException {
  Configuration conf = util.getConfiguration();
  HFile.Reader reader = HFile.createReader(fs, p, new CacheConfig(conf), true, conf);
  HFileScanner scanner = reader.getScanner(false, false);
  scanner.seekTo();
  int count = 0;
  do {
    count++;
  } while (scanner.next());
  reader.close();
  return count;
}
 

/**
 * Method returns the total KVs in given hfile
 * @param fs File System
 * @param p HFile path
 * @return KV count in the given hfile
 * @throws IOException
 */
private static int getKVCountFromHfile(FileSystem fs, Path p) throws IOException {
  Configuration conf = util.getConfiguration();
  HFile.Reader reader = HFile.createReader(fs, p, new CacheConfig(conf), true, conf);
  HFileScanner scanner = reader.getScanner(false, false);
  scanner.seekTo();
  int count = 0;
  do {
    count++;
  } while (scanner.next());
  reader.close();
  return count;
}
 

public MizoHFileIterator(HFileScanner scanner) {
    this.hfileScanner = scanner;
}
 

@Override
public HoplogIterator<byte[], byte[]> scan(long offset, long length)
    throws IOException {
  /**
   * Identifies the first and last key to be scanned based on offset and
   * length. It loads hfile block index and identifies the first hfile block
   * starting after offset. The key of that block is from key for scanner.
   * Similarly it locates first block starting beyond offset + length range.
   * It uses key of that block as the to key for scanner
   */

  // load block indexes in memory
  BlockIndexReader bir = reader.getDataBlockIndexReader();
  int blockCount = bir.getRootBlockCount();
  
  byte[] fromKey = null, toKey = null;

  // find from key
  int i = 0;
  for (; i < blockCount; i++) {
    if (bir.getRootBlockOffset(i) < offset) {
      // hfile block has offset less than this reader's split offset. check
      // the next block
      continue;
    }

    // found the first hfile block starting after offset
    fromKey = bir.getRootBlockKey(i);
    break;
  }

  if (fromKey == null) {
    // seems no block starts after the offset. return no-op scanner
    return new HFileSortedIterator(null, null, false, null, false);
  }
  
  // find to key
  for (; i < blockCount; i++) {
    if (bir.getRootBlockOffset(i) < (offset + length)) {
      // this hfile block lies within the offset+lenght range. check the
      // next block for a higher offset
      continue;
    }

    // found the first block starting beyong offset+length range.
    toKey = bir.getRootBlockKey(i);
    break;
  }

  // from key is included in scan and to key is excluded
  HFileScanner scanner = reader.getScanner(true, false);
  return new HFileSortedIterator(scanner, fromKey, true, toKey, false);
}
 

private void readAllRecords(HFileScanner scanner) throws IOException {
  readAndCheckbytes(scanner, 0, maxRows);
}
 

@Test
public void testWrite() throws Exception {
  Configuration conf = new Configuration();
  HColumnDescriptor columnDescriptor = new HColumnDescriptor();
  // Disable block cache to ensure it reads the actual file content.
  columnDescriptor.setBlockCacheEnabled(false);
  FileSystem fs = FileSystem.get(conf);
  int blockSize = conf.getInt(Constants.HFILE_BLOCKSIZE, 16384);
  final StoreFile.Writer writer =
      new StoreFile.WriterBuilder(conf, new CacheConfig(conf, columnDescriptor), fs, blockSize)
          .withFilePath(new Path(tempFile.toURI()))
          .build();
  /* Create our RecordWriter */
  RecordWriter<BytesWritable, BytesWritable> hfileWriter =
      new HFileRecordWriter(writer);

  List<String> keys = Lists.newArrayList();
  List<String> values = Lists.newArrayList();
  for (int i = 0; i < 100; ++i) {
    String key = String.format("%03d", i);
    String val = "value " + i;
    keys.add(key);
    values.add(val);
    hfileWriter.write(new BytesWritable(key.getBytes()), new BytesWritable(val.getBytes()));
  }
  /* This internally closes the StoreFile.Writer */
  hfileWriter.close(null);

  HFile.Reader reader = HFile.createReader(fs, new Path(tempFile.toURI()),
      new CacheConfig(conf, columnDescriptor));
  HFileScanner scanner = reader.getScanner(false, false, false);
  boolean valid = scanner.seekTo();
  List<String> gotKeys = Lists.newArrayListWithCapacity(keys.size());
  List<String> gotValues = Lists.newArrayListWithCapacity(values.size());
  while(valid) {
    KeyValue keyValue = scanner.getKeyValue();
    gotKeys.add(new String(keyValue.getRow()));
    gotValues.add(new String(keyValue.getValue()));
    valid = scanner.next();
  }
  assertEquals(keys, gotKeys);
  assertEquals(values, gotValues);
  reader.close();
}
 

@Override
public HoplogIterator<byte[], byte[]> scan(long offset, long length)
    throws IOException {
  /**
   * Identifies the first and last key to be scanned based on offset and
   * length. It loads hfile block index and identifies the first hfile block
   * starting after offset. The key of that block is from key for scanner.
   * Similarly it locates first block starting beyond offset + length range.
   * It uses key of that block as the to key for scanner
   */

  // load block indexes in memory
  BlockIndexReader bir = reader.getDataBlockIndexReader();
  int blockCount = bir.getRootBlockCount();
  
  byte[] fromKey = null, toKey = null;

  // find from key
  int i = 0;
  for (; i < blockCount; i++) {
    if (bir.getRootBlockOffset(i) < offset) {
      // hfile block has offset less than this reader's split offset. check
      // the next block
      continue;
    }

    // found the first hfile block starting after offset
    fromKey = bir.getRootBlockKey(i);
    break;
  }

  if (fromKey == null) {
    // seems no block starts after the offset. return no-op scanner
    return new HFileSortedIterator(null, null, false, null, false);
  }
  
  // find to key
  for (; i < blockCount; i++) {
    if (bir.getRootBlockOffset(i) < (offset + length)) {
      // this hfile block lies within the offset+lenght range. check the
      // next block for a higher offset
      continue;
    }

    // found the first block starting beyong offset+length range.
    toKey = bir.getRootBlockKey(i);
    break;
  }

  // from key is included in scan and to key is excluded
  HFileScanner scanner = reader.getScanner(true, false);
  return new HFileSortedIterator(scanner, fromKey, true, toKey, false);
}