/**
 * 
 * @param kv
 * @param isSeek pass true for seek, false for reseek.
 * @return 
 * @throws IOException
 */
public boolean seekOrReseek(Cell cell, boolean isSeek) throws IOException{
    Cell keyToSeek = cell;
    KeyValue splitKeyValue = new KeyValue.KeyOnlyKeyValue(reader.getSplitkey());
    if (reader.isTop()) {
        if(this.comparator.compare(cell, splitKeyValue, true) < 0){
            if(!isSeek && realSeekDone()) {
                return true;
            }
            return seekOrReseekToProperKey(isSeek, keyToSeek);
        }
        keyToSeek = getKeyPresentInHFiles(cell);
        return seekOrReseekToProperKey(isSeek, keyToSeek);
    } else {
        if (this.comparator.compare(cell, splitKeyValue, true) >= 0) {
            close();
            return false;
        }
        if(!isSeek && reader.getRegionInfo().getStartKey().length == 0 && reader.getSplitRow().length > reader.getRegionStartKeyInHFile().length) {
            keyToSeek = getKeyPresentInHFiles(cell);
        }
    }
    return seekOrReseekToProperKey(isSeek, keyToSeek);
}
 

private ResultScanner resultOfScan(List<Cell> cells, Scan scan) throws IOException {
  byte[] startRow = scan.getStartRow();
  byte[] stopRow = scan.getStopRow();

  List<Result> inRangeResults = cells.stream().filter(
      cell -> Bytes.compareTo(startRow, cell.getRowArray()) <= 0
              && Bytes.compareTo(stopRow, cell.getRowArray()) > 0)
      .map(cell -> Result.create(new Cell[] {cell}))
      .collect(toList());

  ResultScanner resultScanner = mock(ResultScanner.class);
  when(resultScanner.iterator()).thenReturn(inRangeResults.iterator());

  if (!inRangeResults.isEmpty()) {
    Result first = inRangeResults.get(0);
    Result[] rest = inRangeResults.subList(1, inRangeResults.size())
        .toArray(new Result[inRangeResults.size()]);
    rest[rest.length - 1] = null; // signal end of scanner
    when(resultScanner.next()).thenReturn(first, rest);
  }

  return resultScanner;
}
 

@Override
public boolean backwardSeek(Cell seekKey) throws IOException {
  if (current == null) {
    return false;
  }
  heap.add(current);
  current = null;

  KeyValueScanner scanner;
  while ((scanner = heap.poll()) != null) {
    Cell topKey = scanner.peek();
    if ((CellUtil.matchingRows(seekKey, topKey) && comparator
        .getComparator().compare(seekKey, topKey) <= 0)
        || comparator.getComparator().compareRows(seekKey, topKey) > 0) {
      heap.add(scanner);
      current = pollRealKV();
      return current != null;
    }
    if (!scanner.backwardSeek(seekKey)) {
      this.scannersForDelayedClose.add(scanner);
    } else {
      heap.add(scanner);
    }
  }
  return false;
}
 

@Override
public boolean checkAndPut(final String tableName, final byte[] rowId, final byte[] family, final byte[] qualifier, final byte[] value, final PutColumn column) throws IOException {
    for (Result result : results.values()) {
        if (Arrays.equals(result.getRow(), rowId)) {
            Cell[] cellArray = result.rawCells();
            for (Cell cell : cellArray) {
                if (Arrays.equals(cell.getFamilyArray(), family) && Arrays.equals(cell.getQualifierArray(), qualifier)) {
                     if (value == null || !Arrays.equals(cell.getValueArray(), value)) {
                         return false;
                     }
                }
            }
        }
    }

    final List<PutColumn> putColumns = new ArrayList<PutColumn>();
    putColumns.add(column);
    put(tableName, rowId, putColumns);
    return true;
}
 

/**
 * Iterate over the list of cells returned from the scan and return a tuple projector for the
 * dynamic columns by parsing the metadata stored for the list of dynamic columns
 * @param result list of cells
 * @param dynCols list of dynamic columns to be populated
 * @param dynColCells list of cells corresponding to dynamic columns to be populated
 * @return The tuple projector corresponding to dynamic columns or null if there are no dynamic
 * columns to process
 * @throws InvalidProtocolBufferException Thrown if there is an error parsing byte[] to protobuf
 */
public static TupleProjector getDynamicColumnsTupleProjector(List<Cell> result,
        List<PColumn> dynCols, List<Cell> dynColCells) throws InvalidProtocolBufferException {
    Set<Pair<ByteBuffer, ByteBuffer>> dynColCellQualifiers = new HashSet<>();
    populateDynColsFromResult(result, dynCols, dynColCellQualifiers);
    if (dynCols.isEmpty()) {
        return null;
    }
    populateDynamicColumnCells(result, dynColCellQualifiers, dynColCells);
    if (dynColCells.isEmpty()) {
        return null;
    }
    KeyValueSchema dynColsSchema = PhoenixRuntime.buildKeyValueSchema(dynCols);
    Expression[] expressions = new Expression[dynCols.size()];
    for (int i = 0; i < dynCols.size(); i++) {
        expressions[i] = new KeyValueColumnExpression(dynCols.get(i));
    }
    return new TupleProjector(dynColsSchema, expressions);
}
 

public void finish(Map<byte[], List<Cell>> familyMaps) {
  if (!isEnable()) {
    return;
  }

  for (Map.Entry<byte[], List<Cell>> e : familyMaps.entrySet()) {
    Store store = this.region.getStore(e.getKey());
    if (store == null || e.getValue() == null) {
      continue;
    }
    if (e.getValue().size() > this.parallelPutToStoreThreadLimitCheckMinColumnCount) {
      AtomicInteger counter = preparePutToStoreMap.get(e.getKey());
      // preparePutToStoreMap will be cleared when changing the configuration, so it may turn
      // into a negative value. It will be not accuracy in a short time, it's a trade-off for
      // performance.
      if (counter != null && counter.decrementAndGet() < 0) {
        counter.incrementAndGet();
      }
    }
  }
}
 

@Test
public void testDeleteVersionMaskingMultiplePuts() throws IOException {
  long now = System.currentTimeMillis();
  KeyValue [] kvs1 = new KeyValue[] {
      create("R1", "cf", "a", now, KeyValue.Type.Put, "dont-care"),
      create("R1", "cf", "a", now, KeyValue.Type.Delete, "dont-care")
  };
  KeyValue [] kvs2 = new KeyValue[] {
      create("R1", "cf", "a", now-500, KeyValue.Type.Put, "dont-care"),
      create("R1", "cf", "a", now-100, KeyValue.Type.Put, "dont-care"),
      create("R1", "cf", "a", now, KeyValue.Type.Put, "dont-care")
  };
  List<KeyValueScanner> scanners = scanFixture(kvs1, kvs2);

  try (StoreScanner scan = new StoreScanner(new Scan().withStartRow(Bytes.toBytes("R1")),
      scanInfo, getCols("a"), scanners)) {
    List<Cell> results = new ArrayList<>();
    // the two put at ts=now will be masked by the 1 delete, and
    // since the scan default returns 1 version we'll return the newest
    // key, which is kvs[2], now-100.
    assertEquals(true, scan.next(results));
    assertEquals(1, results.size());
    assertEquals(kvs2[1], results.get(0));
  }
}
 

protected int binarySearch(final Cell [] kvs,
                           final byte [] family,
                           final byte [] qualifier) {
  byte[] familyNotNull = notNullBytes(family);
  byte[] qualifierNotNull = notNullBytes(qualifier);
  Cell searchTerm =
      PrivateCellUtil.createFirstOnRow(kvs[0].getRowArray(),
          kvs[0].getRowOffset(), kvs[0].getRowLength(),
          familyNotNull, 0, (byte)familyNotNull.length,
          qualifierNotNull, 0, qualifierNotNull.length);

  // pos === ( -(insertion point) - 1)
  int pos = Arrays.binarySearch(kvs, searchTerm, CellComparator.getInstance());
  // 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;
}
 

/**
 * Check if user has given action privilige in cell scope.
 * @param user user name
 * @param table table name
 * @param cell cell to be checked
 * @param action one of action in [Read, Write, Create, Exec, Admin]
 * @return true if user has, false otherwise
 */
public boolean authorizeCell(User user, TableName table, Cell cell, Permission.Action action) {
  try {
    List<Permission> perms = PermissionStorage.getCellPermissionsForUser(user, cell);
    if (LOG.isTraceEnabled()) {
      LOG.trace("Perms for user {} in table {} in cell {}: {}",
        user.getShortName(), table, cell, (perms != null ? perms : ""));
    }
    if (perms != null) {
      for (Permission p: perms) {
        if (p.implies(action)) {
          return true;
        }
      }
    }
  } catch (IOException e) {
    // We failed to parse the KV tag
    LOG.error("Failed parse of ACL tag in cell " + cell);
    // Fall through to check with the table and CF perms we were able
    // to collect regardless
  }
  return false;
}
 

/**
 * @return String
 */
@Override
public String toString() {
  StringBuilder sb = new StringBuilder();
  sb.append("keyvalues=");
  if(isEmpty()) {
    sb.append("NONE");
    return sb.toString();
  }
  sb.append("{");
  boolean moreThanOne = false;
  for(Cell kv : this.cells) {
    if(moreThanOne) {
      sb.append(", ");
    } else {
      moreThanOne = true;
    }
    sb.append(kv.toString());
  }
  sb.append("}");
  return sb.toString();
}
 

private Put setTimeStampOfUnsetValues(final Put put, long ts)
        throws IOException {
  final Put newPut = new Put(put.getRow());
  for (Entry<byte[], List<Cell>> entity : put.getFamilyCellMap().entrySet()) {
    for (Cell cell : entity.getValue()) {
      // If no timestamp was given then use now.
      // This will protect us from a multicluster sumbission
      if (cell.getTimestamp() == HConstants.LATEST_TIMESTAMP) {
        newPut
                .add(cell.getFamily(), cell.getQualifier(), ts, cell.getValue());
      } else {
        newPut.add(cell);
      }
    }
  }
  return newPut;
}
 

Map<byte[], String> readBulkLoadedFiles(String backupId) throws IOException {
  Scan scan = BackupSystemTable.createScanForBulkLoadedFiles(backupId);
  try (Table table = connection.getTable(bulkLoadTableName);
      ResultScanner scanner = table.getScanner(scan)) {
    Result res = null;
    Map<byte[], String> map = new TreeMap<>(Bytes.BYTES_COMPARATOR);
    while ((res = scanner.next()) != null) {
      res.advance();
      byte[] row = CellUtil.cloneRow(res.listCells().get(0));
      for (Cell cell : res.listCells()) {
        if (CellUtil.compareQualifiers(cell, BackupSystemTable.PATH_COL, 0,
          BackupSystemTable.PATH_COL.length) == 0) {
          map.put(row, Bytes.toString(CellUtil.cloneValue(cell)));
        }
      }
    }
    return map;
  }
}
 

/**
 * Gets a list of {@link RegionPruneInfo} for given regions. Returns all regions if the given regions set is null.
 *
 * @param regions a set of regions
 * @return list of {@link RegionPruneInfo}s.
 * @throws IOException when not able to read the data from HBase
 */
public List<RegionPruneInfo> getPruneInfoForRegions(@Nullable SortedSet<byte[]> regions) throws IOException {
  List<RegionPruneInfo> regionPruneInfos = new ArrayList<>();
  try (Table stateTable = stateTableSupplier.get()) {
    byte[] startRow = makeRegionKey(EMPTY_BYTE_ARRAY);
    Scan scan = new Scan(startRow, REGION_KEY_PREFIX_STOP);
    scan.addColumn(FAMILY, PRUNE_UPPER_BOUND_COL);

    try (ResultScanner scanner = stateTable.getScanner(scan)) {
      Result next;
      while ((next = scanner.next()) != null) {
        byte[] region = getRegionFromKey(next.getRow());
        if (regions == null || regions.contains(region)) {
          Cell cell = next.getColumnLatestCell(FAMILY, PRUNE_UPPER_BOUND_COL);
          if (cell != null) {
            byte[] pruneUpperBoundBytes = CellUtil.cloneValue(cell);
            long timestamp = cell.getTimestamp();
            regionPruneInfos.add(new RegionPruneInfo(region, Bytes.toStringBinary(region),
                                                     Bytes.toLong(pruneUpperBoundBytes), timestamp));
          }
        }
      }
    }
  }
  return Collections.unmodifiableList(regionPruneInfos);
}
 

private void internalRun() throws IOException {
  for (long i = 0; i < NUM_TRIES && caughtException.get() == null; i++) {
    MultiVersionConcurrencyControl.WriteEntry w =
        mvcc.begin();

    // Insert the sequence value (i)
    byte[] v = Bytes.toBytes(i);

    KeyValue kv = new KeyValue(row, f, q1, i, v);
    kv.setSequenceId(w.getWriteNumber());
    memstore.add(kv, null);
    mvcc.completeAndWait(w);

    // Assert that we can read back
    KeyValueScanner s = this.memstore.getScanners(mvcc.getReadPoint()).get(0);
    s.seek(kv);

    Cell ret = s.next();
    assertNotNull("Didnt find own write at all", ret);
    assertEquals("Didnt read own writes",
                 kv.getTimestamp(), ret.getTimestamp());
  }
}
 

@Test
public void testGetHRegionInfo() throws IOException {
  assertNull(CatalogFamilyFormat.getRegionInfo(new Result()));

  List<Cell> kvs = new ArrayList<>();
  Result r = Result.create(kvs);
  assertNull(CatalogFamilyFormat.getRegionInfo(r));

  byte[] f = HConstants.CATALOG_FAMILY;
  // Make a key value that doesn't have the expected qualifier.
  kvs.add(new KeyValue(HConstants.EMPTY_BYTE_ARRAY, f, HConstants.SERVER_QUALIFIER, f));
  r = Result.create(kvs);
  assertNull(CatalogFamilyFormat.getRegionInfo(r));
  // Make a key that does not have a regioninfo value.
  kvs.add(new KeyValue(HConstants.EMPTY_BYTE_ARRAY, f, HConstants.REGIONINFO_QUALIFIER, f));
  RegionInfo hri = CatalogFamilyFormat.getRegionInfo(Result.create(kvs));
  assertTrue(hri == null);
  // OK, give it what it expects
  kvs.clear();
  kvs.add(new KeyValue(HConstants.EMPTY_BYTE_ARRAY, f, HConstants.REGIONINFO_QUALIFIER,
    RegionInfo.toByteArray(RegionInfoBuilder.FIRST_META_REGIONINFO)));
  hri = CatalogFamilyFormat.getRegionInfo(Result.create(kvs));
  assertNotNull(hri);
  assertTrue(RegionInfo.COMPARATOR.compare(hri, RegionInfoBuilder.FIRST_META_REGIONINFO) == 0);
}
 

@Override
public String toString() {
  StringBuilder sb = new StringBuilder();
  sb.append("keyvalues=");
  if(this.result == null || this.result.isEmpty()) {
    sb.append("NONE");
    return sb.toString();
  }
  sb.append("{");
  boolean moreThanOne = false;
  for(Cell kv : this.result.listCells()) {
    if(moreThanOne) {
      sb.append(", \n");
    } else {
      moreThanOne = true;
    }
    sb.append(kv.toString()+"/value="+Bytes.toString(kv.getValueArray(), 
      kv.getValueOffset(), kv.getValueLength()));
  }
  sb.append("}\n");
  return sb.toString();
}
 

/**
 * Return the Cells for the specific column.  The Cells are sorted in
 * the {@link CellComparator} order.  That implies the first entry in
 * the list is the most recent column.  If the query (Scan or Get) only
 * requested 1 version the list will contain at most 1 entry.  If the column
 * did not exist in the result set (either the column does not exist
 * or the column was not selected in the query) the list will be empty.
 *
 * Also see getColumnLatest which returns just a Cell
 *
 * @param family the family
 * @param qualifier
 * @return a list of Cells for this column or empty list if the column
 * did not exist in the result set
 */
public List<Cell> getColumnCells(byte [] family, byte [] qualifier) {
  List<Cell> result = new ArrayList<>();

  Cell [] kvs = rawCells();

  if (kvs == null || kvs.length == 0) {
    return result;
  }
  int pos = binarySearch(kvs, family, qualifier);
  if (pos == -1) {
    return result; // cant find it
  }

  for (int i = pos; i < kvs.length; i++) {
    if (CellUtil.matchingColumn(kvs[i], family,qualifier)) {
      result.add(kvs[i]);
    } else {
      break;
    }
  }

  return result;
}
 

private ReturnCode checkNextRow(final Cell cell) {
  final byte[] row = CellUtil.cloneRow(cell);

  System.arraycopy(row, 0, rowCompare, 0, rowCompare.length);

  final int cmp = Bytes.compareTo(rowCompare, nextRow);

  if (cmp < 0) {
    nextCell = CellUtil.createCell(nextRow);
    return ReturnCode.SEEK_NEXT_USING_HINT;
  } else {
    nextCell = null;
    return ReturnCode.INCLUDE;
  }
}
 

@Override
public Cell getCell() {
  if (this.curBlock == null) {
    return null;
  }
  return seeker.getCell();
}
 

private long postAppend(final Entry e, final long elapsedTime) throws IOException {
  long len = 0;
  if (!listeners.isEmpty()) {
    for (Cell cell : e.getEdit().getCells()) {
      len += PrivateCellUtil.estimatedSerializedSizeOf(cell);
    }
    for (WALActionsListener listener : listeners) {
      listener.postAppend(len, elapsedTime, e.getKey(), e.getEdit());
    }
  }
  return len;
}
 

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

public List<Pair<Cell, Cell>> postIncrementBeforeWAL(final Mutation mutation,
    final List<Pair<Cell, Cell>> cellPairs) throws IOException {
  if (this.coprocEnvironments.isEmpty()) {
    return cellPairs;
  }
  return execOperationWithResult(
      new ObserverOperationWithResult<RegionObserver, List<Pair<Cell, Cell>>>(
          regionObserverGetter, cellPairs) {
        @Override
        public List<Pair<Cell, Cell>> call(RegionObserver observer) throws IOException {
          return observer.postIncrementBeforeWAL(this, mutation, getResult());
        }
      });
}
 

protected void assertKey(Cell key, byte [] row, byte [] family, byte [] qualifier,
    byte [] value) {
  assertTrue("Expected row [" + Bytes.toString(row) + "] " +
      "Got row [" + Bytes.toString(CellUtil.cloneRow(key)) +"]",
    equals(row, CellUtil.cloneRow(key)));
  assertTrue("Expected family [" + Bytes.toString(family) + "] " +
      "Got family [" + Bytes.toString(CellUtil.cloneFamily(key)) + "]",
    equals(family, CellUtil.cloneFamily(key)));
  assertTrue("Expected qualifier [" + Bytes.toString(qualifier) + "] " +
      "Got qualifier [" + Bytes.toString(CellUtil.cloneQualifier(key)) + "]",
    equals(qualifier, CellUtil.cloneQualifier(key)));
  assertTrue("Expected value [" + Bytes.toString(value) + "] " +
      "Got value [" + Bytes.toString(CellUtil.cloneValue(key)) + "]",
    equals(value, CellUtil.cloneValue(key)));
}
 

private boolean seekToNextUnfilteredKeyValue() throws IOException {
    while (true) {
        Cell peeked = delegate.peek();
        // no more key values, so we are done
        if (peeked == null) { return false; }

        // filter the peeked value to see if it should be served
        ReturnCode code = filter.filterKeyValue(peeked);
        switch (code) {
        // included, so we are done
        case INCLUDE:
        case INCLUDE_AND_NEXT_COL:
            return true;
            // not included, so we need to go to the next row
        case SKIP:
        case NEXT_COL:
        case NEXT_ROW:
            delegate.next();
            break;
        // use a seek hint to find out where we should go
        case SEEK_NEXT_USING_HINT:
            Cell nextCellHint = filter.getNextCellHint(peeked);
            if(nextCellHint == KeyValue.LOWESTKEY) {
                delegate.next();
            } else {
                delegate.seek(PhoenixKeyValueUtil.maybeCopyCell(nextCellHint));
            }
        }
    }
}
 

private static void assertGet(final HRegion region, byte [] row, byte [] familiy,
    byte[] qualifier, byte[] value) throws IOException {
  // run a get and see if the value matches
  Get get = new Get(row);
  get.addColumn(familiy, qualifier);
  Result result = region.get(get);
  assertEquals(1, result.size());

  Cell kv = result.rawCells()[0];
  byte[] r = CellUtil.cloneValue(kv);
  assertTrue(Bytes.compareTo(r, value) == 0);
}
 

@Test
public void testListIterator() throws Exception {
    EncodedColumnQualiferCellsList list = new EncodedColumnQualiferCellsList(11, 16, FOUR_BYTE_QUALIFIERS);
    Cell[] cells = new Cell[7];
    int i = 0;
    populateListAndArray(list, cells);
    ListIterator<Cell> itr = list.listIterator();
    assertTrue(itr.hasNext());
    
    // test itr.next()
    i = 0;
    while (itr.hasNext()) {
        assertEquals(cells[i++], itr.next());
    }
    
    assertEquals(7, list.size());
    
    // test itr.remove()
    itr = list.listIterator();
    i = 0;
    int numRemoved = 0;
    try {
        itr.remove();
        fail("Remove not allowed till next() is called");
    } catch (IllegalStateException expected) {}
    
    while (itr.hasNext()) {
        assertEquals(cells[i++], itr.next());
        itr.remove();
        numRemoved++;
    }
    assertEquals("Number of elements removed should have been the size of the list", 7, numRemoved);
    assertTrue(list.isEmpty());
}
 

@Override
public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, List<Cell> results) throws IOException {
	LOG.debug("Got preGetOp for row: " + Bytes.toStringBinary(get.getRow()));

	if (Bytes.equals(get.getRow(), ROW_GETDMEO)) {
		KeyValue kv = new KeyValue(get.getRow(), FAMILY, ROW_GETDMEO, Bytes.toBytes("hello,micmiu.com"));
		LOG.debug("coprocess match the row kv: " + kv);
		results.add(kv);
	}
}
 

private ResultCell getResultCell(Cell cell) {
    final ResultCell resultCell = new ResultCell();
    resultCell.setRowArray(cell.getRowArray());
    resultCell.setRowOffset(cell.getRowOffset());
    resultCell.setRowLength(cell.getRowLength());

    resultCell.setFamilyArray(cell.getFamilyArray());
    resultCell.setFamilyOffset(cell.getFamilyOffset());
    resultCell.setFamilyLength(cell.getFamilyLength());

    resultCell.setQualifierArray(cell.getQualifierArray());
    resultCell.setQualifierOffset(cell.getQualifierOffset());
    resultCell.setQualifierLength(cell.getQualifierLength());

    resultCell.setTimestamp(cell.getTimestamp());
    resultCell.setTypeByte(cell.getTypeByte());
    resultCell.setSequenceId(cell.getSequenceId());

    resultCell.setValueArray(cell.getValueArray());
    resultCell.setValueOffset(cell.getValueOffset());
    resultCell.setValueLength(cell.getValueLength());

    resultCell.setTagsArray(cell.getTagsArray());
    resultCell.setTagsOffset(cell.getTagsOffset());
    resultCell.setTagsLength(cell.getTagsLength());
    return resultCell;
}
 

@Test
public void testReseeksWithOneColumnMiltipleTimestamp() throws IOException {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  byte [] FAMILY = Bytes.toBytes("event_log");
  byte [][] FAMILIES = new byte[][] { FAMILY };

  // create table; set versions to max...
  Table ht = TEST_UTIL.createTable(tableName, FAMILIES, Integer.MAX_VALUE);

  Integer[] putRows = new Integer[] {1, 3, 5, 7};
  Integer[] putColumns = new Integer[] { 1, 3, 5};
  Long[] putTimestamps = new Long[] {1L, 2L, 3L, 4L, 5L};

  Integer[] scanRows = new Integer[] {3, 5};
  Integer[] scanColumns = new Integer[] {3};
  Long[] scanTimestamps = new Long[] {3L, 4L};
  int scanMaxVersions = 2;

  put(ht, FAMILY, putRows, putColumns, putTimestamps);

  TEST_UTIL.flush(tableName);

  ResultScanner scanner = scan(ht, FAMILY, scanRows, scanColumns,
      scanTimestamps, scanMaxVersions);

  Cell[] kvs;

  kvs = scanner.next().rawCells();
  assertEquals(2, kvs.length);
  checkOneCell(kvs[0], FAMILY, 3, 3, 4);
  checkOneCell(kvs[1], FAMILY, 3, 3, 3);
  kvs = scanner.next().rawCells();
  assertEquals(2, kvs.length);
  checkOneCell(kvs[0], FAMILY, 5, 3, 4);
  checkOneCell(kvs[1], FAMILY, 5, 3, 3);

  ht.close();
}
 

@Override
public Cell getValue(int index) {
    if (index != 0 || cell == null) {
        throw new IndexOutOfBoundsException(Integer.toString(index));
    }
    return cell;
}