@Test
  @SuppressWarnings("checkstyle:MethodLength")
  public void testDuplicateVersions() throws Exception {
    final TableName tableName = name.getTableName();

    long[] STAMPS = makeStamps(20);
    byte[][] VALUES = makeNAscii(VALUE, 20);

    try (Table ht = TEST_UTIL.createTable(tableName, FAMILY, 10)) {

      // Insert 4 versions of same column
      Put put = new Put(ROW);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[4], VALUES[4]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[5], VALUES[5]);
      ht.put(put);

      // Verify we can get each one properly
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[4], VALUES[4]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[5], VALUES[5]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[4], VALUES[4]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[5], VALUES[5]);

      // Verify we don't accidentally get others
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[0]);
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[3]);
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[6]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[0]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[3]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[6]);

      // Ensure maxVersions in query is respected
      Get get = new Get(ROW);
      get.addColumn(FAMILY, QUALIFIER);
      get.readVersions(2);
      Result result = ht.get(get);
      assertNResult(result, ROW, FAMILY, QUALIFIER, new long[] { STAMPS[4], STAMPS[5] },
        new byte[][] { VALUES[4], VALUES[5] }, 0, 1);

      Scan scan = new Scan().withStartRow(ROW);
      scan.addColumn(FAMILY, QUALIFIER);
      scan.readVersions(2);
      result = getSingleScanResult(ht, scan);
      assertNResult(result, ROW, FAMILY, QUALIFIER, new long[] { STAMPS[4], STAMPS[5] },
        new byte[][] { VALUES[4], VALUES[5] }, 0, 1);

      // Flush and redo

      TEST_UTIL.flush();

      // Verify we can get each one properly
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[4], VALUES[4]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[5], VALUES[5]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[4], VALUES[4]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[5], VALUES[5]);

      // Verify we don't accidentally get others
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[0]);
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[3]);
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[6]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[0]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[3]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[6]);

      // Ensure maxVersions in query is respected
      get = new Get(ROW);
      get.addColumn(FAMILY, QUALIFIER);
      get.readVersions(2);
      result = ht.get(get);
      assertNResult(result, ROW, FAMILY, QUALIFIER, new long[] { STAMPS[4], STAMPS[5] },
        new byte[][] { VALUES[4], VALUES[5] }, 0, 1);

      scan = new Scan().withStartRow(ROW);
      scan.addColumn(FAMILY, QUALIFIER);
      scan.readVersions(2);
      result = getSingleScanResult(ht, scan);
      assertNResult(result, ROW, FAMILY, QUALIFIER, new long[] { STAMPS[4], STAMPS[5] },
        new byte[][] { VALUES[4], VALUES[5] }, 0, 1);

      // Add some memstore and retest

      // Insert 4 more versions of same column and a dupe
      put = new Put(ROW);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[3], VALUES[3]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[4], VALUES[14]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[6], VALUES[6]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[7], VALUES[7]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[8], VALUES[8]);
      ht.put(put);

      // Ensure maxVersions in query is respected
      get = new Get(ROW);
      get.addColumn(FAMILY, QUALIFIER);
      get.readVersions(7);
      result = ht.get(get);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[2], STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[7], STAMPS[8] },
        new byte[][] { VALUES[2], VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[7],
          VALUES[8] },
        0, 6);

      scan = new Scan().withStartRow(ROW);
      scan.addColumn(FAMILY, QUALIFIER);
      scan.readVersions(7);
      result = getSingleScanResult(ht, scan);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[2], STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[7], STAMPS[8] },
        new byte[][] { VALUES[2], VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[7],
          VALUES[8] },
        0, 6);

      get = new Get(ROW);
      get.readVersions(7);
      result = ht.get(get);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[2], STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[7], STAMPS[8] },
        new byte[][] { VALUES[2], VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[7],
          VALUES[8] },
        0, 6);

      scan = new Scan().withStartRow(ROW);
      scan.readVersions(7);
      result = getSingleScanResult(ht, scan);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[2], STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[7], STAMPS[8] },
        new byte[][] { VALUES[2], VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[7],
          VALUES[8] },
        0, 6);

      // Verify we can get each one properly
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[4], VALUES[14]);
      getVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[7], VALUES[7]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[1], VALUES[1]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[2], VALUES[2]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[4], VALUES[14]);
      scanVersionAndVerify(ht, ROW, FAMILY, QUALIFIER, STAMPS[7], VALUES[7]);

      // Verify we don't accidentally get others
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[0]);
      getVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[9]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[0]);
      scanVersionAndVerifyMissing(ht, ROW, FAMILY, QUALIFIER, STAMPS[9]);

      // Ensure maxVersions of table is respected

      TEST_UTIL.flush();

      // Insert 4 more versions of same column and a dupe
      put = new Put(ROW);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[9], VALUES[9]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[11], VALUES[11]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[13], VALUES[13]);
      put.addColumn(FAMILY, QUALIFIER, STAMPS[15], VALUES[15]);
      ht.put(put);

      get = new Get(ROW);
      get.addColumn(FAMILY, QUALIFIER);
      get.readVersions(Integer.MAX_VALUE);
      result = ht.get(get);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[7], STAMPS[8], STAMPS[9],
          STAMPS[11], STAMPS[13], STAMPS[15] },
        new byte[][] { VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[7], VALUES[8], VALUES[9],
          VALUES[11], VALUES[13], VALUES[15] },
        0, 9);

      scan = new Scan().withStartRow(ROW);
      scan.addColumn(FAMILY, QUALIFIER);
      scan.readVersions(Integer.MAX_VALUE);
      result = getSingleScanResult(ht, scan);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[7], STAMPS[8], STAMPS[9],
          STAMPS[11], STAMPS[13], STAMPS[15] },
        new byte[][] { VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[7], VALUES[8], VALUES[9],
          VALUES[11], VALUES[13], VALUES[15] },
        0, 9);

      // Delete a version in the memstore and a version in a storefile
      Delete delete = new Delete(ROW);
      delete.addColumn(FAMILY, QUALIFIER, STAMPS[11]);
      delete.addColumn(FAMILY, QUALIFIER, STAMPS[7]);
      ht.delete(delete);

      // Test that it's gone
      get = new Get(ROW);
      get.addColumn(FAMILY, QUALIFIER);
      get.readVersions(Integer.MAX_VALUE);
      result = ht.get(get);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[1], STAMPS[2], STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[8],
          STAMPS[9], STAMPS[13], STAMPS[15] },
        new byte[][] { VALUES[1], VALUES[2], VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[8],
          VALUES[9], VALUES[13], VALUES[15] },
        0, 9);

      scan = new Scan().withStartRow(ROW);
      scan.addColumn(FAMILY, QUALIFIER);
      scan.readVersions(Integer.MAX_VALUE);
      result = getSingleScanResult(ht, scan);
      assertNResult(result, ROW, FAMILY, QUALIFIER,
        new long[] { STAMPS[1], STAMPS[2], STAMPS[3], STAMPS[4], STAMPS[5], STAMPS[6], STAMPS[8],
          STAMPS[9], STAMPS[13], STAMPS[15] },
        new byte[][] { VALUES[1], VALUES[2], VALUES[3], VALUES[14], VALUES[5], VALUES[6], VALUES[8],
          VALUES[9], VALUES[13], VALUES[15] },
        0, 9);
    }
  }
HBASE-52 Add a means of scanning over all versions
hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide4.java