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

/**
 * @param conf
 * @param family             Name of this store's column family
 * @param minVersions        Store's MIN_VERSIONS setting
 * @param maxVersions        Store's VERSIONS setting
 * @param ttl                Store's TTL (in ms)
 * @param timeToPurgeDeletes duration in ms after which a delete marker can
 *                           be purged during a major compaction.
 * @param keepDeletedCells   Store's keepDeletedCells setting
 * @param comparator         The store's comparator
 */
public ScanInfo(final Configuration conf, final byte[] family, final int minVersions,
    final int maxVersions, final long ttl, final KeepDeletedCells keepDeletedCells,
    final long timeToPurgeDeletes, final KVComparator comparator) {
  this.family = family;
  this.minVersions = minVersions;
  this.maxVersions = maxVersions;
  this.ttl = ttl;
  this.keepDeletedCells = keepDeletedCells;
  this.timeToPurgeDeletes = timeToPurgeDeletes;
  this.comparator = comparator;
  this.tableMaxRowSize =
      conf.getLong(HConstants.TABLE_MAX_ROWSIZE_KEY, HConstants.TABLE_MAX_ROWSIZE_DEFAULT);
  this.usePread = conf.getBoolean("hbase.storescanner.use.pread", false);
  long perHeartbeat = conf.getLong(StoreScanner.HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK,
      StoreScanner.DEFAULT_HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK);
  this.cellsPerTimeoutCheck = perHeartbeat > 0 ?
      perHeartbeat :
      StoreScanner.DEFAULT_HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK;
  this.parallelSeekEnabled =
      conf.getBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, false);
  this.conf = conf;
}

/**
 * The ExplicitColumnTracker does not support "raw" scanning.
 */
@Test
public void testRawScanWithColumns() throws Exception {
  HTableDescriptor htd = hbu.createTableDescriptor(name.getMethodName(), 0, 3,
      HConstants.FOREVER, KeepDeletedCells.TRUE);
  HRegion region = hbu.createLocalHRegion(htd, null, null);

  Scan s = new Scan();
  s.setRaw(true);
  s.setMaxVersions();
  s.addColumn(c0, c0);

  try {
    region.getScanner(s);
    fail("raw scanner with columns should have failed");
  } catch (org.apache.hadoop.hbase.DoNotRetryIOException dnre) {
    // ok!
  }

  HRegion.closeHRegion(region);
}

/**
 * The ExplicitColumnTracker does not support "raw" scanning.
 */
@Test
public void testRawScanWithColumns() throws Exception {
  HTableDescriptor htd = hbu.createTableDescriptor(name.getMethodName(), 0, 3,
      HConstants.FOREVER, KeepDeletedCells.TRUE);
  HRegion region = hbu.createLocalHRegion(htd, null, null);

  Scan s = new Scan();
  s.setRaw(true);
  s.setMaxVersions();
  s.addColumn(c0, c0);

  try {
    region.getScanner(s);
    fail("raw scanner with columns should have failed");
  } catch (org.apache.hadoop.hbase.DoNotRetryIOException dnre) {
    // ok!
  }

  HRegion.closeHRegion(region);
}

private void testDropDeletes(
    byte[] from, byte[] to, byte[][] rows, MatchCode... expected) throws IOException {
  long now = EnvironmentEdgeManager.currentTime();
  // Set time to purge deletes to negative value to avoid it ever happening.
  ScanInfo scanInfo = new ScanInfo(fam2, 0, 1, ttl, KeepDeletedCells.FALSE, -1L, rowComparator);
  NavigableSet<byte[]> cols = get.getFamilyMap().get(fam2);

  ScanQueryMatcher qm = new ScanQueryMatcher(scan, scanInfo, cols, Long.MAX_VALUE,
      HConstants.OLDEST_TIMESTAMP, HConstants.OLDEST_TIMESTAMP, now, from, to, null);
  List<ScanQueryMatcher.MatchCode> actual =
      new ArrayList<ScanQueryMatcher.MatchCode>(rows.length);
  byte[] prevRow = null;
  for (byte[] row : rows) {
    if (prevRow == null || !Bytes.equals(prevRow, row)) {
      qm.setRow(row, 0, (short)row.length);
      prevRow = row;
    }
    actual.add(qm.match(new KeyValue(row, fam2, null, now, Type.Delete)));
  }

  assertEquals(expected.length, actual.size());
  for (int i = 0; i < expected.length; i++) {
    if (PRINT) System.out.println("expected " + expected[i] + ", actual " + actual.get(i));
    assertEquals(expected[i], actual.get(i));
  }
}

protected final MatchCode tryDropDelete(Cell cell) {
  long timestamp = cell.getTimestamp();
  // If it is not the time to drop the delete marker, just return
  if (timeToPurgeDeletes > 0 && now - timestamp <= timeToPurgeDeletes) {
    return MatchCode.INCLUDE;
  }
  if (keepDeletedCells == KeepDeletedCells.TRUE
      || (keepDeletedCells == KeepDeletedCells.TTL && timestamp >= oldestUnexpiredTS)) {
    // If keepDeletedCell is true, or the delete marker is not expired yet, we should include it
    // in version counting to see if we can drop it. The only exception is that, we can make
    // sure that no put is older than this delete marker. And under this situation, all later
    // cells of this column(must be delete markers) can be skipped.
    if (timestamp < earliestPutTs) {
      return columns.getNextRowOrNextColumn(cell);
    } else {
      return null;
    }
  } else {
    return MatchCode.SKIP;
  }
}

private ScanInfo(byte[] family, int minVersions, int maxVersions, long ttl,
    KeepDeletedCells keepDeletedCells, long timeToPurgeDeletes, CellComparator comparator,
    long tableMaxRowSize, boolean usePread, long cellsPerTimeoutCheck,
    boolean parallelSeekEnabled, long preadMaxBytes, boolean newVersionBehavior) {
  this.family = family;
  this.minVersions = minVersions;
  this.maxVersions = maxVersions;
  this.ttl = ttl;
  this.keepDeletedCells = keepDeletedCells;
  this.timeToPurgeDeletes = timeToPurgeDeletes;
  this.comparator = comparator;
  this.tableMaxRowSize = tableMaxRowSize;
  this.usePread = usePread;
  this.cellsPerTimeoutCheck = cellsPerTimeoutCheck;
  this.parallelSeekEnabled = parallelSeekEnabled;
  this.preadMaxBytes = preadMaxBytes;
  this.newVersionBehavior = newVersionBehavior;
}

/**
 * The ExplicitColumnTracker does not support "raw" scanning.
 */
@Test
public void testRawScanWithColumns() throws Exception {
  HTableDescriptor htd = hbu.createTableDescriptor(name.getMethodName(), 0, 3,
      HConstants.FOREVER, KeepDeletedCells.TRUE);
  Region region = hbu.createLocalHRegion(htd, null, null);

  Scan s = new Scan();
  s.setRaw(true);
  s.setMaxVersions();
  s.addColumn(c0, c0);

  try {
    region.getScanner(s);
    fail("raw scanner with columns should have failed");
  } catch (org.apache.hadoop.hbase.DoNotRetryIOException dnre) {
    // ok!
  }

  HBaseTestingUtility.closeRegionAndWAL(region);
}

private void testDropDeletes(byte[] from, byte[] to, byte[][] rows, MatchCode... expected)
    throws IOException {
  long now = EnvironmentEdgeManager.currentTime();
  // Set time to purge deletes to negative value to avoid it ever happening.
  ScanInfo scanInfo = new ScanInfo(this.conf, fam2, 0, 1, ttl, KeepDeletedCells.FALSE,
      HConstants.DEFAULT_BLOCKSIZE, -1L, rowComparator, false);

  CompactionScanQueryMatcher qm = CompactionScanQueryMatcher.create(scanInfo,
    ScanType.COMPACT_RETAIN_DELETES, Long.MAX_VALUE, HConstants.OLDEST_TIMESTAMP,
    HConstants.OLDEST_TIMESTAMP, now, from, to, null);
  List<ScanQueryMatcher.MatchCode> actual = new ArrayList<>(rows.length);
  byte[] prevRow = null;
  for (byte[] row : rows) {
    if (prevRow == null || !Bytes.equals(prevRow, row)) {
      qm.setToNewRow(KeyValueUtil.createFirstOnRow(row));
      prevRow = row;
    }
    actual.add(qm.match(new KeyValue(row, fam2, null, now, Type.Delete)));
  }

  assertEquals(expected.length, actual.size());
  for (int i = 0; i < expected.length; i++) {
    LOG.debug("expected " + expected[i] + ", actual " + actual.get(i));
    assertEquals(expected[i], actual.get(i));
  }
}

/**
 * Ensure that expired delete family markers don't override valid puts
 */
@Test
public void testExpiredDeleteFamily() throws Exception {
  long now = System.currentTimeMillis();
  KeyValue[] kvs = new KeyValue[] {
      new KeyValue(Bytes.toBytes("R1"), Bytes.toBytes("cf"), null, now-1000,
          KeyValue.Type.DeleteFamily),
      create("R1", "cf", "a", now-10, KeyValue.Type.Put,
          "dont-care"),
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scan = new Scan();
  scan.readVersions(1);
  // scanner with ttl equal to 500
  ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, 1, 500, KeepDeletedCells.FALSE,
      HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
  try (StoreScanner scanner = new StoreScanner(scan, scanInfo, null, scanners)) {
    List<Cell> results = new ArrayList<>();
    assertEquals(true, scanner.next(results));
    assertEquals(1, results.size());
    assertEquals(kvs[1], results.get(0));
    results.clear();

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

private void _testMatch_ExplicitColumns(Scan scan, List<MatchCode> expected) throws IOException {
  long now = EnvironmentEdgeManager.currentTime();
  // 2,4,5
  ScanQueryMatcher qm = new ScanQueryMatcher(scan, new ScanInfo(this.conf, fam2,
      0, 1, ttl, KeepDeletedCells.FALSE, 0, rowComparator), get.getFamilyMap().get(fam2),
      now - ttl, now);

  List<KeyValue> memstore = new ArrayList<KeyValue>();
  memstore.add(new KeyValue(row1, fam2, col1, 1, data));
  memstore.add(new KeyValue(row1, fam2, col2, 1, data));
  memstore.add(new KeyValue(row1, fam2, col3, 1, data));
  memstore.add(new KeyValue(row1, fam2, col4, 1, data));
  memstore.add(new KeyValue(row1, fam2, col5, 1, data));

  memstore.add(new KeyValue(row2, fam1, col1, data));

  List<ScanQueryMatcher.MatchCode> actual = new ArrayList<ScanQueryMatcher.MatchCode>();
  KeyValue k = memstore.get(0);
  qm.setRow(k.getRowArray(), k.getRowOffset(), k.getRowLength());

  for (KeyValue kv : memstore){
    actual.add(qm.match(kv));
  }

  assertEquals(expected.size(), actual.size());
  for(int i=0; i< expected.size(); i++){
    assertEquals(expected.get(i), actual.get(i));
    if(PRINT){
      System.out.println("expected "+expected.get(i)+
          ", actual " +actual.get(i));
    }
  }
}

private void testDropDeletes(
    byte[] from, byte[] to, byte[][] rows, MatchCode... expected) throws IOException {
  long now = EnvironmentEdgeManager.currentTime();
  // Set time to purge deletes to negative value to avoid it ever happening.
  ScanInfo scanInfo =
    new ScanInfo(this.conf, fam2, 0, 1, ttl, KeepDeletedCells.FALSE, -1L, rowComparator);
  NavigableSet<byte[]> cols = get.getFamilyMap().get(fam2);

  ScanQueryMatcher qm = new ScanQueryMatcher(scan, scanInfo, cols, Long.MAX_VALUE,
      HConstants.OLDEST_TIMESTAMP, HConstants.OLDEST_TIMESTAMP, now, from, to, null);
  List<ScanQueryMatcher.MatchCode> actual =
      new ArrayList<ScanQueryMatcher.MatchCode>(rows.length);
  byte[] prevRow = null;
  for (byte[] row : rows) {
    if (prevRow == null || !Bytes.equals(prevRow, row)) {
      qm.setRow(row, 0, (short)row.length);
      prevRow = row;
    }
    actual.add(qm.match(new KeyValue(row, fam2, null, now, Type.Delete)));
  }

  assertEquals(expected.length, actual.size());
  for (int i = 0; i < expected.length; i++) {
    if (PRINT) System.out.println("expected " + expected[i] + ", actual " + actual.get(i));
    assertEquals(expected[i], actual.get(i));
  }
}

public void testWildCardTtlScan() throws IOException {
  long now = System.currentTimeMillis();
  KeyValue [] kvs = new KeyValue[] {
      KeyValueTestUtil.create("R1", "cf", "a", now-1000, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "b", now-10, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "c", now-200, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "d", now-10000, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "a", now, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "b", now-10, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "c", now-200, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "c", now-1000, KeyValue.Type.Put, "dont-care")
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scan = new Scan();
  scan.setMaxVersions(1);
  ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, 1, 500, KeepDeletedCells.FALSE, 0,
      KeyValue.COMPARATOR);
  ScanType scanType = ScanType.USER_SCAN;
  StoreScanner scanner =
    new StoreScanner(scan, scanInfo, scanType,
        null, scanners);

  List<Cell> results = new ArrayList<Cell>();
  assertEquals(true, scanner.next(results));
  assertEquals(2, results.size());
  assertEquals(kvs[1], results.get(0));
  assertEquals(kvs[2], results.get(1));
  results.clear();

  assertEquals(true, scanner.next(results));
  assertEquals(3, results.size());
  assertEquals(kvs[4], results.get(0));
  assertEquals(kvs[5], results.get(1));
  assertEquals(kvs[6], results.get(2));
  results.clear();

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

/**
 * Ensure that expired delete family markers don't override valid puts
 */
public void testExpiredDeleteFamily() throws Exception {
  long now = System.currentTimeMillis();
  KeyValue [] kvs = new KeyValue[] {
      new KeyValue(Bytes.toBytes("R1"), Bytes.toBytes("cf"), null, now-1000,
          KeyValue.Type.DeleteFamily),
      KeyValueTestUtil.create("R1", "cf", "a", now-10, KeyValue.Type.Put,
          "dont-care"),
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scan = new Scan();
  scan.setMaxVersions(1);
  // scanner with ttl equal to 500
  ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, 1, 500, KeepDeletedCells.FALSE, 0,
      KeyValue.COMPARATOR);
  ScanType scanType = ScanType.USER_SCAN;
  StoreScanner scanner =
      new StoreScanner(scan, scanInfo, scanType, null, scanners);

  List<Cell> results = new ArrayList<Cell>();
  assertEquals(true, scanner.next(results));
  assertEquals(1, results.size());
  assertEquals(kvs[1], results.get(0));
  results.clear();

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

private static HColumnDescriptor createColumnFamily() {
    return new HColumnDescriptor(CF_NAME)
            .setMaxVersions(1)
            .setBlockCacheEnabled(true)
            .setBloomFilterType(BloomType.ROW)
            .setCompressionType(DEFAULT_COMPRESSION_ALGORITHM)
            .setDataBlockEncoding(DEFAULT_DATABLOCK_ENCODING)
            .setCacheBloomsOnWrite(true)
            .setCacheDataOnWrite(true)
            .setCacheIndexesOnWrite(true)
            .setKeepDeletedCells(KeepDeletedCells.FALSE)
            .setValue(HTableDescriptor.MAX_FILESIZE, REGION_MAX_FILESIZE)
            .setValue(HTableDescriptor.SPLIT_POLICY, REGION_SPLIT_POLICY);
}

/**
 * @param family Name of this store's column family
 * @param minVersions Store's MIN_VERSIONS setting
 * @param maxVersions Store's VERSIONS setting
 * @param ttl Store's TTL (in ms)
 * @param timeToPurgeDeletes duration in ms after which a delete marker can
 *        be purged during a major compaction.
 * @param keepDeletedCells Store's keepDeletedCells setting
 * @param comparator The store's comparator
 */
public ScanInfo(final byte[] family, final int minVersions, final int maxVersions,
    final long ttl, final KeepDeletedCells keepDeletedCells, final long timeToPurgeDeletes,
    final KVComparator comparator) {
  this.family = family;
  this.minVersions = minVersions;
  this.maxVersions = maxVersions;
  this.ttl = ttl;
  this.keepDeletedCells = keepDeletedCells;
  this.timeToPurgeDeletes = timeToPurgeDeletes;
  this.comparator = comparator;
}

/**
 * Verify that delete markers are removed from an otherwise empty store.
 */
@Test
public void testDeleteMarkerExpirationEmptyStore() throws Exception {
  HTableDescriptor htd = hbu.createTableDescriptor(name.getMethodName(), 0, 1,
      HConstants.FOREVER, KeepDeletedCells.TRUE);
  HRegion region = hbu.createLocalHRegion(htd, null, null);

  long ts = EnvironmentEdgeManager.currentTime();

  Delete d = new Delete(T1, ts);
  d.deleteColumns(c0, c0, ts);
  region.delete(d);

  d = new Delete(T1, ts);
  d.deleteFamily(c0);
  region.delete(d);

  d = new Delete(T1, ts);
  d.deleteColumn(c0, c0, ts+1);
  region.delete(d);

  d = new Delete(T1, ts);
  d.deleteColumn(c0, c0, ts+2);
  region.delete(d);

  // 1 family marker, 1 column marker, 2 version markers
  assertEquals(4, countDeleteMarkers(region));

  // neither flush nor minor compaction removes any marker
  region.flushcache();
  assertEquals(4, countDeleteMarkers(region));
  region.compactStores(false);
  assertEquals(4, countDeleteMarkers(region));

  // major compaction removes all, since there are no puts they affect
  region.compactStores(true);
  assertEquals(0, countDeleteMarkers(region));

  HRegion.closeHRegion(region);
}

private void _testMatch_ExplicitColumns(Scan scan, List<MatchCode> expected) throws IOException {
  long now = EnvironmentEdgeManager.currentTime();
  // 2,4,5
  ScanQueryMatcher qm = new ScanQueryMatcher(scan, new ScanInfo(fam2,
      0, 1, ttl, KeepDeletedCells.FALSE, 0, rowComparator), get.getFamilyMap().get(fam2),
      now - ttl, now);

  List<KeyValue> memstore = new ArrayList<KeyValue>();
  memstore.add(new KeyValue(row1, fam2, col1, 1, data));
  memstore.add(new KeyValue(row1, fam2, col2, 1, data));
  memstore.add(new KeyValue(row1, fam2, col3, 1, data));
  memstore.add(new KeyValue(row1, fam2, col4, 1, data));
  memstore.add(new KeyValue(row1, fam2, col5, 1, data));

  memstore.add(new KeyValue(row2, fam1, col1, data));

  List<ScanQueryMatcher.MatchCode> actual = new ArrayList<ScanQueryMatcher.MatchCode>();
  KeyValue k = memstore.get(0);
  qm.setRow(k.getRowArray(), k.getRowOffset(), k.getRowLength());

  for (KeyValue kv : memstore){
    actual.add(qm.match(kv));
  }

  assertEquals(expected.size(), actual.size());
  for(int i=0; i< expected.size(); i++){
    assertEquals(expected.get(i), actual.get(i));
    if(PRINT){
      System.out.println("expected "+expected.get(i)+
          ", actual " +actual.get(i));
    }
  }
}

public void testWildCardTtlScan() throws IOException {
  long now = System.currentTimeMillis();
  KeyValue [] kvs = new KeyValue[] {
      KeyValueTestUtil.create("R1", "cf", "a", now-1000, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "b", now-10, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "c", now-200, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R1", "cf", "d", now-10000, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "a", now, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "b", now-10, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "c", now-200, KeyValue.Type.Put, "dont-care"),
      KeyValueTestUtil.create("R2", "cf", "c", now-1000, KeyValue.Type.Put, "dont-care")
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scan = new Scan();
  scan.setMaxVersions(1);
  ScanInfo scanInfo = new ScanInfo(CF, 0, 1, 500, KeepDeletedCells.FALSE, 0,
      KeyValue.COMPARATOR);
  ScanType scanType = ScanType.USER_SCAN;
  StoreScanner scanner =
    new StoreScanner(scan, scanInfo, scanType,
        null, scanners);

  List<Cell> results = new ArrayList<Cell>();
  assertEquals(true, scanner.next(results));
  assertEquals(2, results.size());
  assertEquals(kvs[1], results.get(0));
  assertEquals(kvs[2], results.get(1));
  results.clear();

  assertEquals(true, scanner.next(results));
  assertEquals(3, results.size());
  assertEquals(kvs[4], results.get(0));
  assertEquals(kvs[5], results.get(1));
  assertEquals(kvs[6], results.get(2));
  results.clear();

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

/**
 * Ensure that expired delete family markers don't override valid puts
 */
public void testExpiredDeleteFamily() throws Exception {
  long now = System.currentTimeMillis();
  KeyValue [] kvs = new KeyValue[] {
      new KeyValue(Bytes.toBytes("R1"), Bytes.toBytes("cf"), null, now-1000,
          KeyValue.Type.DeleteFamily),
      KeyValueTestUtil.create("R1", "cf", "a", now-10, KeyValue.Type.Put,
          "dont-care"),
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scan = new Scan();
  scan.setMaxVersions(1);
  // scanner with ttl equal to 500
  ScanInfo scanInfo = new ScanInfo(CF, 0, 1, 500, KeepDeletedCells.FALSE, 0,
      KeyValue.COMPARATOR);
  ScanType scanType = ScanType.USER_SCAN;
  StoreScanner scanner =
      new StoreScanner(scan, scanInfo, scanType, null, scanners);

  List<Cell> results = new ArrayList<Cell>();
  assertEquals(true, scanner.next(results));
  assertEquals(1, results.size());
  assertEquals(kvs[1], results.get(0));
  results.clear();

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

public static void main(String[] args) throws Exception {
    String quorum = "192.168.0.30,192.168.0.31,192.168.0.32";
    //quorum = "192.168.8.191,192.168.1.192,192.168.1.193";
    int port = 2181;
    String znode = "/hyperbase1";
    HBaseConnPool connPool = new HBaseClientManager(quorum, port, znode);
    HBaseDDLHandler ddlHandler = new HBaseDDLHandler(connPool);

    String tableName = "demo_test";
    System.out.println("=============================== : delete");
    ddlHandler.deleteTable(tableName);

    String columnFamily = "cf";
    System.out.println("=============================== : create");
    ddlHandler.createTable(tableName, columnFamily, "cf2");

    System.out.println("=============================== : desc");
    HBaseUtils.printTableInfo(ddlHandler.getTable(tableName));
    System.out.println("=============================== : alter");
    HBaseAdmin admin = new HBaseAdmin(connPool.getConn());
    admin.disableTable(tableName);
    HTableInterface htable = ddlHandler.getTable(tableName);
    HTableDescriptor tableDesc = admin.getTableDescriptor(htable.getTableName());
    tableDesc.removeFamily(Bytes.toBytes("cf2"));
    HColumnDescriptor newhcd = new HColumnDescriptor("cf3");
    newhcd.setMaxVersions(2);
    newhcd.setKeepDeletedCells(KeepDeletedCells.TRUE);
    tableDesc.addFamily(newhcd);

    admin.modifyTable(tableName, tableDesc);
    admin.enableTable(tableName);
    admin.close();

    System.out.println("=============================== : desc");
    HBaseUtils.printTableInfo(ddlHandler.getTable(tableName));
    System.out.println("=============================== : delete");
    ddlHandler.deleteTable(tableName);

    connPool.closeConn();
}

protected NormalUserScanQueryMatcher(Scan scan, ScanInfo scanInfo, ColumnTracker columns,
    boolean hasNullColumn, DeleteTracker deletes, long oldestUnexpiredTS, long now) {
  super(scan, scanInfo, columns, hasNullColumn, oldestUnexpiredTS, now);
  this.deletes = deletes;
  this.get = scan.isGetScan();
  this.seePastDeleteMarkers = scanInfo.getKeepDeletedCells() != KeepDeletedCells.FALSE;
}

protected final void trackDelete(Cell cell) {
  // If keepDeletedCells is true, then we only remove cells by versions or TTL during
  // compaction, so we do not need to track delete here.
  // If keepDeletedCells is TTL and the delete marker is expired, then we can make sure that the
  // minVerions is larger than 0(otherwise we will just return at preCheck). So here we still
  // need to track the delete marker to see if it masks some cells.
  if (keepDeletedCells == KeepDeletedCells.FALSE
      || (keepDeletedCells == KeepDeletedCells.TTL && cell.getTimestamp() < oldestUnexpiredTS)) {
    deletes.add(cell);
  }
}

/**
 * Verify that delete markers are removed from an otherwise empty store.
 */
@Test
public void testDeleteMarkerExpirationEmptyStore() throws Exception {
  HTableDescriptor htd = hbu.createTableDescriptor(name.getMethodName(), 0, 1,
      HConstants.FOREVER, KeepDeletedCells.TRUE);
  HRegion region = hbu.createLocalHRegion(htd, null, null);

  long ts = EnvironmentEdgeManager.currentTime();

  Delete d = new Delete(T1, ts);
  d.addColumns(c0, c0, ts);
  region.delete(d);

  d = new Delete(T1, ts);
  d.addFamily(c0);
  region.delete(d);

  d = new Delete(T1, ts);
  d.addColumn(c0, c0, ts+1);
  region.delete(d);

  d = new Delete(T1, ts);
  d.addColumn(c0, c0, ts+2);
  region.delete(d);

  // 1 family marker, 1 column marker, 2 version markers
  assertEquals(4, countDeleteMarkers(region));

  // neither flush nor minor compaction removes any marker
  region.flush(true);
  assertEquals(4, countDeleteMarkers(region));
  region.compact(false);
  assertEquals(4, countDeleteMarkers(region));

  // major compaction removes all, since there are no puts they affect
  region.compact(true);
  assertEquals(0, countDeleteMarkers(region));

  HBaseTestingUtility.closeRegionAndWAL(region);
}

/**
 * This is a cryptic test. It is checking that we don't include a fake cell, one that has a
 * timestamp of {@link HConstants#OLDEST_TIMESTAMP}. See HBASE-16074 for background.
 * @throws IOException
 */
@Test
public void testNeverIncludeFakeCell() throws IOException {
  long now = EnvironmentEdgeManager.currentTime();
  // Do with fam2 which has a col2 qualifier.
  UserScanQueryMatcher qm = UserScanQueryMatcher.create(scan,
    new ScanInfo(this.conf, fam2, 10, 1, ttl, KeepDeletedCells.FALSE,
        HConstants.DEFAULT_BLOCKSIZE, 0, rowComparator, false),
    get.getFamilyMap().get(fam2), now - ttl, now, null);
  Cell kv = new KeyValue(row1, fam2, col2, 1, data);
  Cell cell = PrivateCellUtil.createLastOnRowCol(kv);
  qm.setToNewRow(kv);
  MatchCode code = qm.match(cell);
  assertFalse(code.compareTo(MatchCode.SEEK_NEXT_COL) != 0);
}

/**
 * Verify that {@link ScanQueryMatcher} only skips expired KeyValue instances and does not exit
 * early from the row (skipping later non-expired KeyValues). This version mimics a Get with
 * explicitly specified column qualifiers.
 * @throws IOException
 */
@Test
public void testMatch_ExpiredExplicit() throws IOException {

  long testTTL = 1000;
  MatchCode[] expected = new MatchCode[] { ScanQueryMatcher.MatchCode.SEEK_NEXT_COL,
      ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_COL,
      ScanQueryMatcher.MatchCode.SEEK_NEXT_COL,
      ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_COL,
      ScanQueryMatcher.MatchCode.SEEK_NEXT_ROW, ScanQueryMatcher.MatchCode.DONE };

  long now = EnvironmentEdgeManager.currentTime();
  UserScanQueryMatcher qm = UserScanQueryMatcher.create(scan,
    new ScanInfo(this.conf, fam2, 0, 1, testTTL, KeepDeletedCells.FALSE,
        HConstants.DEFAULT_BLOCKSIZE, 0, rowComparator, false),
    get.getFamilyMap().get(fam2), now - testTTL, now, null);

  KeyValue[] kvs = new KeyValue[] { new KeyValue(row1, fam2, col1, now - 100, data),
      new KeyValue(row1, fam2, col2, now - 50, data),
      new KeyValue(row1, fam2, col3, now - 5000, data),
      new KeyValue(row1, fam2, col4, now - 500, data),
      new KeyValue(row1, fam2, col5, now - 10000, data),
      new KeyValue(row2, fam1, col1, now - 10, data) };

  KeyValue k = kvs[0];
  qm.setToNewRow(k);

  List<MatchCode> actual = new ArrayList<>(kvs.length);
  for (KeyValue kv : kvs) {
    actual.add(qm.match(kv));
  }

  assertEquals(expected.length, actual.size());
  for (int i = 0; i < expected.length; i++) {
    LOG.debug("expected " + expected[i] + ", actual " + actual.get(i));
    assertEquals(expected[i], actual.get(i));
  }
}

/**
 * Verify that {@link ScanQueryMatcher} only skips expired KeyValue instances and does not exit
 * early from the row (skipping later non-expired KeyValues). This version mimics a Get with
 * wildcard-inferred column qualifiers.
 * @throws IOException
 */
@Test
public void testMatch_ExpiredWildcard() throws IOException {

  long testTTL = 1000;
  MatchCode[] expected =
      new MatchCode[] { ScanQueryMatcher.MatchCode.INCLUDE, ScanQueryMatcher.MatchCode.INCLUDE,
        ScanQueryMatcher.MatchCode.SEEK_NEXT_COL, ScanQueryMatcher.MatchCode.INCLUDE,
        ScanQueryMatcher.MatchCode.SEEK_NEXT_COL, ScanQueryMatcher.MatchCode.DONE };

  long now = EnvironmentEdgeManager.currentTime();
  UserScanQueryMatcher qm =
      UserScanQueryMatcher.create(scan,
        new ScanInfo(this.conf, fam2, 0, 1, testTTL, KeepDeletedCells.FALSE,
            HConstants.DEFAULT_BLOCKSIZE, 0, rowComparator, false),
        null, now - testTTL, now, null);

  KeyValue[] kvs = new KeyValue[] { new KeyValue(row1, fam2, col1, now - 100, data),
      new KeyValue(row1, fam2, col2, now - 50, data),
      new KeyValue(row1, fam2, col3, now - 5000, data),
      new KeyValue(row1, fam2, col4, now - 500, data),
      new KeyValue(row1, fam2, col5, now - 10000, data),
      new KeyValue(row2, fam1, col1, now - 10, data) };
  KeyValue k = kvs[0];
  qm.setToNewRow(k);

  List<ScanQueryMatcher.MatchCode> actual = new ArrayList<>(kvs.length);
  for (KeyValue kv : kvs) {
    actual.add(qm.match(kv));
  }

  assertEquals(expected.length, actual.size());
  for (int i = 0; i < expected.length; i++) {
    LOG.debug("expected " + expected[i] + ", actual " + actual.get(i));
    assertEquals(expected[i], actual.get(i));
  }
}

@Test
public void testMatchWhenFilterReturnsIncludeAndSeekNextRow() throws IOException {
  List<MatchCode> expected = new ArrayList<>();
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_ROW);
  expected.add(ScanQueryMatcher.MatchCode.DONE);

  Scan scanWithFilter = new Scan(scan).setFilter(new AlwaysIncludeAndSeekNextRowFilter());

  long now = EnvironmentEdgeManager.currentTime();

  // scan with column 2,4,5
  UserScanQueryMatcher qm = UserScanQueryMatcher.create(
    scanWithFilter, new ScanInfo(this.conf, fam2, 0, 1, ttl, KeepDeletedCells.FALSE,
        HConstants.DEFAULT_BLOCKSIZE, 0, rowComparator, false),
    get.getFamilyMap().get(fam2), now - ttl, now, null);

  List<KeyValue> memstore = new ArrayList<>();
  // ColumnTracker will return INCLUDE_AND_SEEK_NEXT_COL , and filter will return
  // INCLUDE_AND_SEEK_NEXT_ROW, so final match code will be INCLUDE_AND_SEEK_NEXT_ROW.
  memstore.add(new KeyValue(row1, fam2, col2, 1, data));
  memstore.add(new KeyValue(row2, fam1, col1, data));

  List<ScanQueryMatcher.MatchCode> actual = new ArrayList<>(memstore.size());
  KeyValue k = memstore.get(0);
  qm.setToNewRow(k);

  for (KeyValue kv : memstore) {
    actual.add(qm.match(kv));
  }

  assertEquals(expected.size(), actual.size());
  for (int i = 0; i < expected.size(); i++) {
    LOG.debug("expected " + expected.get(i) + ", actual " + actual.get(i));
    assertEquals(expected.get(i), actual.get(i));
  }
}

/**
 * Here is the unit test for UserScanQueryMatcher#mergeFilterResponse: the match code may be
 * changed to SEEK_NEXT_COL or INCLUDE_AND_SEEK_NEXT_COL after merging with filterResponse, even
 * if the passed match code is neither SEEK_NEXT_COL nor INCLUDE_AND_SEEK_NEXT_COL. In that case,
 * we need to make sure that the ColumnTracker has been switched to the next column. <br/>
 * An effective test way is: we only need to check the cell from getKeyForNextColumn(). because
 * that as long as the UserScanQueryMatcher returns SEEK_NEXT_COL or INCLUDE_AND_SEEK_NEXT_COL,
 * UserScanQueryMatcher#getKeyForNextColumn should return an cell whose column is larger than the
 * current cell's.
 */
@Test
public void testMergeFilterResponseCase2() throws Exception {
  List<MatchCode> expected = new ArrayList<>();
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.SEEK_NEXT_COL);

  Scan scanWithFilter = new Scan(scan).setFilter(new AlwaysIncludeFilter()).readVersions(3);

  long now = EnvironmentEdgeManager.currentTime();

  // scan with column 2,4,5, the family with maxVersion = 5
  UserScanQueryMatcher qm = UserScanQueryMatcher.create(
    scanWithFilter, new ScanInfo(this.conf, fam2, 0, 5, ttl, KeepDeletedCells.FALSE,
        HConstants.DEFAULT_BLOCKSIZE, 0, rowComparator, false),
    get.getFamilyMap().get(fam2), now - ttl, now, null);

  List<KeyValue> memstore = new ArrayList<>();

  memstore.add(new KeyValue(row1, fam1, col2, 1, data)); // match code will be INCLUDE
  memstore.add(new KeyValue(row1, fam1, col2, 2, data)); // match code will be INCLUDE
  memstore.add(new KeyValue(row1, fam1, col2, 3, data)); // match code will be INCLUDE
  memstore.add(new KeyValue(row1, fam1, col2, 4, data)); // match code will be SEEK_NEXT_COL

  KeyValue k = memstore.get(0);
  qm.setToNewRow(k);

  for (int i = 0; i < memstore.size(); i++) {
    assertEquals(expected.get(i), qm.match(memstore.get(i)));
  }

  // For last cell, the query matcher will return SEEK_NEXT_COL, and the
  // ColumnTracker will skip to the next column, which is col4.
  Cell lastCell = memstore.get(memstore.size() - 1);
  Cell nextCell = qm.getKeyForNextColumn(lastCell);
  assertArrayEquals(nextCell.getQualifierArray(), col4);
}

@Test
public void testWildCardTtlScan() throws IOException {
  long now = System.currentTimeMillis();
  KeyValue [] kvs = new KeyValue[] {
      create("R1", "cf", "a", now-1000, KeyValue.Type.Put, "dont-care"),
      create("R1", "cf", "b", now-10, KeyValue.Type.Put, "dont-care"),
      create("R1", "cf", "c", now-200, KeyValue.Type.Put, "dont-care"),
      create("R1", "cf", "d", now-10000, KeyValue.Type.Put, "dont-care"),
      create("R2", "cf", "a", now, KeyValue.Type.Put, "dont-care"),
      create("R2", "cf", "b", now-10, KeyValue.Type.Put, "dont-care"),
      create("R2", "cf", "c", now-200, KeyValue.Type.Put, "dont-care"),
      create("R2", "cf", "c", now-1000, KeyValue.Type.Put, "dont-care")
  };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  Scan scan = new Scan();
  scan.readVersions(1);
  ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, 1, 500, KeepDeletedCells.FALSE,
      HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
  try (StoreScanner scanner = new StoreScanner(scan, scanInfo, null, scanners)) {
    List<Cell> results = new ArrayList<>();
    assertEquals(true, scanner.next(results));
    assertEquals(2, results.size());
    assertEquals(kvs[1], results.get(0));
    assertEquals(kvs[2], results.get(1));
    results.clear();

    assertEquals(true, scanner.next(results));
    assertEquals(3, results.size());
    assertEquals(kvs[4], results.get(0));
    assertEquals(kvs[5], results.get(1));
    assertEquals(kvs[6], results.get(2));
    results.clear();

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

@Test
public void testPreadNotEnabledForCompactionStoreScanners() throws Exception {
  long now = System.currentTimeMillis();
  KeyValue[] kvs = new KeyValue[] {
      new KeyValue(Bytes.toBytes("R1"), Bytes.toBytes("cf"), null, now - 1000,
          KeyValue.Type.DeleteFamily),
      create("R1", "cf", "a", now - 10, KeyValue.Type.Put, "dont-care"), };
  List<KeyValueScanner> scanners = scanFixture(kvs);
  ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, 1, 500, KeepDeletedCells.FALSE,
      HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
  try (StoreScanner storeScanner = new StoreScanner(scanInfo, OptionalInt.empty(),
      ScanType.COMPACT_RETAIN_DELETES, scanners)) {
    assertFalse(storeScanner.isScanUsePread());
  }
}

/**
 * Construct a QueryMatcher for a scan
 * @param scan
 * @param scanInfo The store's immutable scan info
 * @param columns
 * @param scanType Type of the scan
 * @param earliestPutTs Earliest put seen in any of the store files.
 * @param oldestUnexpiredTS the oldest timestamp we are interested in,
 *  based on TTL
 * @param regionCoprocessorHost 
 * @throws IOException 
 */
public ScanQueryMatcher(Scan scan, ScanInfo scanInfo, NavigableSet<byte[]> columns,
    ScanType scanType, long readPointToUse, long earliestPutTs, long oldestUnexpiredTS,
    long now, RegionCoprocessorHost regionCoprocessorHost) throws IOException {
  TimeRange timeRange = scan.getColumnFamilyTimeRange().get(scanInfo.getFamily());
  if (timeRange == null) {
    this.tr = scan.getTimeRange();
  } else {
    this.tr = timeRange;
  }
  this.rowComparator = scanInfo.getComparator();
  this.regionCoprocessorHost = regionCoprocessorHost;
  this.deletes =  instantiateDeleteTracker();
  this.stopRow = scan.getStopRow();
  this.startKey = KeyValueUtil.createFirstDeleteFamilyOnRow(scan.getStartRow(),
      scanInfo.getFamily());
  this.filter = scan.getFilter();
  this.earliestPutTs = earliestPutTs;
  this.oldestUnexpiredTS = oldestUnexpiredTS;
  this.now = now;

  this.maxReadPointToTrackVersions = readPointToUse;
  this.timeToPurgeDeletes = scanInfo.getTimeToPurgeDeletes();
  this.ttl = oldestUnexpiredTS;

  /* how to deal with deletes */
  this.isUserScan = scanType == ScanType.USER_SCAN;
  // keep deleted cells: if compaction or raw scan
  this.keepDeletedCells = scan.isRaw() ? KeepDeletedCells.TRUE :
    isUserScan ? KeepDeletedCells.FALSE : scanInfo.getKeepDeletedCells();
  // retain deletes: if minor compaction or raw scanisDone
  this.retainDeletesInOutput = scanType == ScanType.COMPACT_RETAIN_DELETES || scan.isRaw();
  // seePastDeleteMarker: user initiated scans
  this.seePastDeleteMarkers =
      scanInfo.getKeepDeletedCells() != KeepDeletedCells.FALSE && isUserScan;

  int maxVersions =
      scan.isRaw() ? scan.getMaxVersions() : Math.min(scan.getMaxVersions(),
        scanInfo.getMaxVersions());

  // Single branch to deal with two types of reads (columns vs all in family)
  if (columns == null || columns.size() == 0) {
    // there is always a null column in the wildcard column query.
    hasNullColumn = true;

    // use a specialized scan for wildcard column tracker.
    this.columns = new ScanWildcardColumnTracker(
        scanInfo.getMinVersions(), maxVersions, oldestUnexpiredTS);
  } else {
    // whether there is null column in the explicit column query
    hasNullColumn = (columns.first().length == 0);

    // We can share the ExplicitColumnTracker, diff is we reset
    // between rows, not between storefiles.
    this.columns = new ExplicitColumnTracker(columns, scanInfo.getMinVersions(), maxVersions,
        oldestUnexpiredTS);
  }
  this.isReversed = scan.isReversed();
}

KeepDeletedCells getKeepDeletedCells() {
  return keepDeletedCells;
}

/**
 * basic verification of existing behavior
 */
@Test
public void testWithoutKeepingDeletes() throws Exception {
  // KEEP_DELETED_CELLS is NOT enabled
  HTableDescriptor htd = hbu.createTableDescriptor(name.getMethodName(), 0, 3,
      HConstants.FOREVER, KeepDeletedCells.FALSE);
  HRegion region = hbu.createLocalHRegion(htd, null, null);

  long ts = EnvironmentEdgeManager.currentTime();
  Put p = new Put(T1, ts);
  p.add(c0, c0, T1);
  region.put(p);

  Get gOne = new Get(T1);
  gOne.setMaxVersions();
  gOne.setTimeRange(0L, ts + 1);
  Result rOne = region.get(gOne);
  assertFalse(rOne.isEmpty());


  Delete d = new Delete(T1, ts+2);
  d.deleteColumn(c0, c0, ts);
  region.delete(d);

  // "past" get does not see rows behind delete marker
  Get g = new Get(T1);
  g.setMaxVersions();
  g.setTimeRange(0L, ts+1);
  Result r = region.get(g);
  assertTrue(r.isEmpty());

  // "past" scan does not see rows behind delete marker
  Scan s = new Scan();
  s.setMaxVersions();
  s.setTimeRange(0L, ts+1);
  InternalScanner scanner = region.getScanner(s);
  List<Cell> kvs = new ArrayList<Cell>();
  while (scanner.next(kvs))
    ;
  assertTrue(kvs.isEmpty());

  // flushing and minor compaction keep delete markers
  region.flush(true);
  region.compact(false);
  assertEquals(1, countDeleteMarkers(region));
  region.compact(true);
  // major compaction deleted it
  assertEquals(0, countDeleteMarkers(region));

  HRegion.closeHRegion(region);
}

/** Test getNextRow from memstore
 * @throws InterruptedException
 */
public void testGetNextRow() throws Exception {
  addRows(this.memstore);
  // Add more versions to make it a little more interesting.
  Thread.sleep(1);
  addRows(this.memstore);
  Cell closestToEmpty = this.memstore.getNextRow(KeyValue.LOWESTKEY);
  assertTrue(KeyValue.COMPARATOR.compareRows(closestToEmpty,
    new KeyValue(Bytes.toBytes(0), System.currentTimeMillis())) == 0);
  for (int i = 0; i < ROW_COUNT; i++) {
    Cell nr = this.memstore.getNextRow(new KeyValue(Bytes.toBytes(i),
      System.currentTimeMillis()));
    if (i + 1 == ROW_COUNT) {
      assertEquals(nr, null);
    } else {
      assertTrue(KeyValue.COMPARATOR.compareRows(nr,
        new KeyValue(Bytes.toBytes(i + 1), System.currentTimeMillis())) == 0);
    }
  }
  //starting from each row, validate results should contain the starting row
  Configuration conf = HBaseConfiguration.create();
  for (int startRowId = 0; startRowId < ROW_COUNT; startRowId++) {
    ScanInfo scanInfo = new ScanInfo(conf, FAMILY, 0, 1, Integer.MAX_VALUE,
      KeepDeletedCells.FALSE, 0, this.memstore.comparator);
    ScanType scanType = ScanType.USER_SCAN;
    InternalScanner scanner = new StoreScanner(new Scan(
        Bytes.toBytes(startRowId)), scanInfo, scanType, null,
        memstore.getScanners(0));
    List<Cell> results = new ArrayList<Cell>();
    for (int i = 0; scanner.next(results); i++) {
      int rowId = startRowId + i;
      Cell left = results.get(0);
      byte[] row1 = Bytes.toBytes(rowId);
      assertTrue(
          "Row name",
          KeyValue.COMPARATOR.compareRows(left.getRowArray(), left.getRowOffset(),
              (int) left.getRowLength(), row1, 0, row1.length) == 0);
      assertEquals("Count of columns", QUALIFIER_COUNT, results.size());
      List<Cell> row = new ArrayList<Cell>();
      for (Cell kv : results) {
        row.add(kv);
      }
      isExpectedRowWithoutTimestamps(rowId, row);
      // Clear out set.  Otherwise row results accumulate.
      results.clear();
    }
  }
}

public void testMatch_Wildcard()
throws IOException {
  //Moving up from the Tracker by using Gets and List<KeyValue> instead
  //of just byte []

  //Expected result
  List<MatchCode> expected = new ArrayList<ScanQueryMatcher.MatchCode>();
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.INCLUDE);
  expected.add(ScanQueryMatcher.MatchCode.DONE);

  long now = EnvironmentEdgeManager.currentTime();
  ScanQueryMatcher qm = new ScanQueryMatcher(scan, new ScanInfo(this.conf, fam2,
      0, 1, ttl, KeepDeletedCells.FALSE, 0, rowComparator), null,
      now - ttl, now);

  List<KeyValue> memstore = new ArrayList<KeyValue>();
  memstore.add(new KeyValue(row1, fam2, col1, 1, data));
  memstore.add(new KeyValue(row1, fam2, col2, 1, data));
  memstore.add(new KeyValue(row1, fam2, col3, 1, data));
  memstore.add(new KeyValue(row1, fam2, col4, 1, data));
  memstore.add(new KeyValue(row1, fam2, col5, 1, data));
  memstore.add(new KeyValue(row2, fam1, col1, 1, data));

  List<ScanQueryMatcher.MatchCode> actual = new ArrayList<ScanQueryMatcher.MatchCode>();

  KeyValue k = memstore.get(0);
  qm.setRow(k.getRowArray(), k.getRowOffset(), k.getRowLength());

  for(KeyValue kv : memstore) {
    actual.add(qm.match(kv));
  }

  assertEquals(expected.size(), actual.size());
  for(int i=0; i< expected.size(); i++){
    assertEquals(expected.get(i), actual.get(i));
    if(PRINT){
      System.out.println("expected "+expected.get(i)+
          ", actual " +actual.get(i));
    }
  }
}