hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestReversibleScanners.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.regionserver;
  19
  20 import static org.junit.Assert.assertEquals;
  21 import static org.junit.Assert.assertFalse;
  22 import static org.junit.Assert.assertTrue;
  23
  24 import java.io.IOException;
  25 import java.util.ArrayList;
  26 import java.util.Collections;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.NavigableSet;
  30 import org.apache.hadoop.fs.FileSystem;
  31 import org.apache.hadoop.fs.Path;
  32 import org.apache.hadoop.hbase.Cell;
  33 import org.apache.hadoop.hbase.CellComparatorImpl;
  34 import org.apache.hadoop.hbase.CompareOperator;
  35 import org.apache.hadoop.hbase.HBaseClassTestRule;
  36 import org.apache.hadoop.hbase.HBaseTestingUtil;
  37 import org.apache.hadoop.hbase.HConstants;
  38 import org.apache.hadoop.hbase.KeepDeletedCells;
  39 import org.apache.hadoop.hbase.KeyValue;
  40 import org.apache.hadoop.hbase.KeyValueUtil;
  41 import org.apache.hadoop.hbase.TableName;
  42 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
  43 import org.apache.hadoop.hbase.client.Put;
  44 import org.apache.hadoop.hbase.client.Result;
  45 import org.apache.hadoop.hbase.client.Scan;
  46 import org.apache.hadoop.hbase.client.TableDescriptor;
  47 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
  48 import org.apache.hadoop.hbase.filter.Filter;
  49 import org.apache.hadoop.hbase.filter.FilterList;
  50 import org.apache.hadoop.hbase.filter.FilterList.Operator;
  51 import org.apache.hadoop.hbase.filter.PageFilter;
  52 import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
  53 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
  54 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
  55 import org.apache.hadoop.hbase.io.hfile.HFileContext;
  56 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
  57 import org.apache.hadoop.hbase.testclassification.MediumTests;
  58 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
  59 import org.apache.hadoop.hbase.util.Bytes;
  60 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
  61 import org.apache.hadoop.hbase.util.Pair;
  62 import org.junit.BeforeClass;
  63 import org.junit.ClassRule;
  64 import org.junit.Rule;
  65 import org.junit.Test;
  66 import org.junit.experimental.categories.Category;
  67 import org.junit.rules.TestName;
  68 import org.slf4j.Logger;
  69 import org.slf4j.LoggerFactory;
  70
  71 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
  72
  73 /**
  74  * Test cases against ReversibleKeyValueScanner
  75  */
  76 @Category({RegionServerTests.class, MediumTests.class})
  77 public class TestReversibleScanners {
  78
  79   @ClassRule
  80   public static final HBaseClassTestRule CLASS_RULE =
  81       HBaseClassTestRule.forClass(TestReversibleScanners.class);
  82
  83   private static final Logger LOG = LoggerFactory.getLogger(TestReversibleScanners.class);
  84   HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
  85
  86   private static byte[] FAMILYNAME = Bytes.toBytes("testCf");
  87   private static long TS = EnvironmentEdgeManager.currentTime();
  88   private static int MAXMVCC = 7;
  89   private static byte[] ROW = Bytes.toBytes("testRow");
  90   private static final int ROWSIZE = 200;
  91   private static byte[][] ROWS = makeN(ROW, ROWSIZE);
  92   private static byte[] QUAL = Bytes.toBytes("testQual");
  93   private static final int QUALSIZE = 5;
  94   private static byte[][] QUALS = makeN(QUAL, QUALSIZE);
  95   private static byte[] VALUE = Bytes.toBytes("testValue");
  96   private static final int VALUESIZE = 3;
  97   private static byte[][] VALUES = makeN(VALUE, VALUESIZE);
  98
  99   @Rule
 100   public TestName name = new TestName();
 101
 102   @BeforeClass
 103   public static void setUp() {
 104     ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0,
 105       0, null, MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT);
 106   }
 107   @Test
 108   public void testReversibleStoreFileScanner() throws IOException {
 109     FileSystem fs = TEST_UTIL.getTestFileSystem();
 110     Path hfilePath = new Path(new Path(
 111         TEST_UTIL.getDataTestDir("testReversibleStoreFileScanner"),
 112         "regionname"), "familyname");
 113     CacheConfig cacheConf = new CacheConfig(TEST_UTIL.getConfiguration());
 114     for (DataBlockEncoding encoding : DataBlockEncoding.values()) {
 115       HFileContextBuilder hcBuilder = new HFileContextBuilder();
 116       hcBuilder.withBlockSize(2 * 1024);
 117       hcBuilder.withDataBlockEncoding(encoding);
 118       HFileContext hFileContext = hcBuilder.build();
 119       StoreFileWriter writer = new StoreFileWriter.Builder(
 120           TEST_UTIL.getConfiguration(), cacheConf, fs).withOutputDir(hfilePath)
 121           .withFileContext(hFileContext).build();
 122       writeStoreFile(writer);
 123
 124       HStoreFile sf = new HStoreFile(fs, writer.getPath(), TEST_UTIL.getConfiguration(), cacheConf,
 125           BloomType.NONE, true);
 126
 127       List<StoreFileScanner> scanners = StoreFileScanner
 128           .getScannersForStoreFiles(Collections.singletonList(sf),
 129               false, true, false, false, Long.MAX_VALUE);
 130       StoreFileScanner scanner = scanners.get(0);
 131       seekTestOfReversibleKeyValueScanner(scanner);
 132       for (int readPoint = 0; readPoint < MAXMVCC; readPoint++) {
 133         LOG.info("Setting read point to " + readPoint);
 134         scanners = StoreFileScanner.getScannersForStoreFiles(
 135             Collections.singletonList(sf), false, true, false, false, readPoint);
 136         seekTestOfReversibleKeyValueScannerWithMVCC(scanners, readPoint);
 137       }
 138     }
 139
 140   }
 141
 142   @Test
 143   public void testReversibleMemstoreScanner() throws IOException {
 144     MemStore memstore = new DefaultMemStore();
 145     writeMemstore(memstore);
 146     List<KeyValueScanner> scanners = memstore.getScanners(Long.MAX_VALUE);
 147     seekTestOfReversibleKeyValueScanner(scanners.get(0));
 148     for (int readPoint = 0; readPoint < MAXMVCC; readPoint++) {
 149       LOG.info("Setting read point to " + readPoint);
 150       scanners = memstore.getScanners(readPoint);
 151       seekTestOfReversibleKeyValueScannerWithMVCC(scanners, readPoint);
 152     }
 153
 154   }
 155
 156   @Test
 157   public void testReversibleKeyValueHeap() throws IOException {
 158     // write data to one memstore and two store files
 159     FileSystem fs = TEST_UTIL.getTestFileSystem();
 160     Path hfilePath = new Path(new Path(
 161         TEST_UTIL.getDataTestDir("testReversibleKeyValueHeap"), "regionname"),
 162         "familyname");
 163     CacheConfig cacheConf = new CacheConfig(TEST_UTIL.getConfiguration());
 164     HFileContextBuilder hcBuilder = new HFileContextBuilder();
 165     hcBuilder.withBlockSize(2 * 1024);
 166     HFileContext hFileContext = hcBuilder.build();
 167     StoreFileWriter writer1 = new StoreFileWriter.Builder(
 168         TEST_UTIL.getConfiguration(), cacheConf, fs).withOutputDir(
 169         hfilePath).withFileContext(hFileContext).build();
 170     StoreFileWriter writer2 = new StoreFileWriter.Builder(
 171         TEST_UTIL.getConfiguration(), cacheConf, fs).withOutputDir(
 172         hfilePath).withFileContext(hFileContext).build();
 173
 174     MemStore memstore = new DefaultMemStore();
 175     writeMemstoreAndStoreFiles(memstore, new StoreFileWriter[] { writer1,
 176         writer2 });
 177
 178     HStoreFile sf1 = new HStoreFile(fs, writer1.getPath(), TEST_UTIL.getConfiguration(), cacheConf,
 179         BloomType.NONE, true);
 180
 181     HStoreFile sf2 = new HStoreFile(fs, writer2.getPath(), TEST_UTIL.getConfiguration(), cacheConf,
 182         BloomType.NONE, true);
 183     /**
 184      * Test without MVCC
 185      */
 186     int startRowNum = ROWSIZE / 2;
 187     ReversedKeyValueHeap kvHeap = getReversibleKeyValueHeap(memstore, sf1, sf2,
 188         ROWS[startRowNum], MAXMVCC);
 189     internalTestSeekAndNextForReversibleKeyValueHeap(kvHeap, startRowNum);
 190
 191     startRowNum = ROWSIZE - 1;
 192     kvHeap = getReversibleKeyValueHeap(memstore, sf1, sf2,
 193         HConstants.EMPTY_START_ROW, MAXMVCC);
 194     internalTestSeekAndNextForReversibleKeyValueHeap(kvHeap, startRowNum);
 195
 196     /**
 197      * Test with MVCC
 198      */
 199     for (int readPoint = 0; readPoint < MAXMVCC; readPoint++) {
 200       LOG.info("Setting read point to " + readPoint);
 201       startRowNum = ROWSIZE - 1;
 202       kvHeap = getReversibleKeyValueHeap(memstore, sf1, sf2,
 203           HConstants.EMPTY_START_ROW, readPoint);
 204       for (int i = startRowNum; i >= 0; i--) {
 205         if (i - 2 < 0) break;
 206         i = i - 2;
 207         kvHeap.seekToPreviousRow(KeyValueUtil.createFirstOnRow(ROWS[i + 1]));
 208         Pair<Integer, Integer> nextReadableNum = getNextReadableNumWithBackwardScan(
 209             i, 0, readPoint);
 210         if (nextReadableNum == null) break;
 211         KeyValue expecedKey = makeKV(nextReadableNum.getFirst(),
 212             nextReadableNum.getSecond());
 213         assertEquals(expecedKey, kvHeap.peek());
 214         i = nextReadableNum.getFirst();
 215         int qualNum = nextReadableNum.getSecond();
 216         if (qualNum + 1 < QUALSIZE) {
 217           kvHeap.backwardSeek(makeKV(i, qualNum + 1));
 218           nextReadableNum = getNextReadableNumWithBackwardScan(i, qualNum + 1,
 219               readPoint);
 220           if (nextReadableNum == null) break;
 221           expecedKey = makeKV(nextReadableNum.getFirst(),
 222               nextReadableNum.getSecond());
 223           assertEquals(expecedKey, kvHeap.peek());
 224           i = nextReadableNum.getFirst();
 225           qualNum = nextReadableNum.getSecond();
 226         }
 227
 228         kvHeap.next();
 229
 230         if (qualNum + 1 >= QUALSIZE) {
 231           nextReadableNum = getNextReadableNumWithBackwardScan(i - 1, 0,
 232               readPoint);
 233         } else {
 234           nextReadableNum = getNextReadableNumWithBackwardScan(i, qualNum + 1,
 235               readPoint);
 236         }
 237         if (nextReadableNum == null) break;
 238         expecedKey = makeKV(nextReadableNum.getFirst(),
 239             nextReadableNum.getSecond());
 240         assertEquals(expecedKey, kvHeap.peek());
 241         i = nextReadableNum.getFirst();
 242       }
 243     }
 244   }
 245
 246   @Test
 247   public void testReversibleStoreScanner() throws IOException {
 248     // write data to one memstore and two store files
 249     FileSystem fs = TEST_UTIL.getTestFileSystem();
 250     Path hfilePath = new Path(new Path(
 251         TEST_UTIL.getDataTestDir("testReversibleStoreScanner"), "regionname"),
 252         "familyname");
 253     CacheConfig cacheConf = new CacheConfig(TEST_UTIL.getConfiguration());
 254     HFileContextBuilder hcBuilder = new HFileContextBuilder();
 255     hcBuilder.withBlockSize(2 * 1024);
 256     HFileContext hFileContext = hcBuilder.build();
 257     StoreFileWriter writer1 = new StoreFileWriter.Builder(
 258         TEST_UTIL.getConfiguration(), cacheConf, fs).withOutputDir(
 259         hfilePath).withFileContext(hFileContext).build();
 260     StoreFileWriter writer2 = new StoreFileWriter.Builder(
 261         TEST_UTIL.getConfiguration(), cacheConf, fs).withOutputDir(
 262         hfilePath).withFileContext(hFileContext).build();
 263
 264     MemStore memstore = new DefaultMemStore();
 265     writeMemstoreAndStoreFiles(memstore, new StoreFileWriter[] { writer1,
 266         writer2 });
 267
 268     HStoreFile sf1 = new HStoreFile(fs, writer1.getPath(), TEST_UTIL.getConfiguration(), cacheConf,
 269         BloomType.NONE, true);
 270
 271     HStoreFile sf2 = new HStoreFile(fs, writer2.getPath(), TEST_UTIL.getConfiguration(), cacheConf,
 272         BloomType.NONE, true);
 273
 274     ScanInfo scanInfo =
 275         new ScanInfo(TEST_UTIL.getConfiguration(), FAMILYNAME, 0, Integer.MAX_VALUE, Long.MAX_VALUE,
 276             KeepDeletedCells.FALSE, HConstants.DEFAULT_BLOCKSIZE, 0, CellComparatorImpl.COMPARATOR, false);
 277
 278     // Case 1.Test a full reversed scan
 279     Scan scan = new Scan();
 280     scan.setReversed(true);
 281     StoreScanner storeScanner =
 282         getReversibleStoreScanner(memstore, sf1, sf2, scan, scanInfo, MAXMVCC);
 283     verifyCountAndOrder(storeScanner, QUALSIZE * ROWSIZE, ROWSIZE, false);
 284
 285     // Case 2.Test reversed scan with a specified start row
 286     int startRowNum = ROWSIZE / 2;
 287     byte[] startRow = ROWS[startRowNum];
 288     scan.withStartRow(startRow);
 289     storeScanner = getReversibleStoreScanner(memstore, sf1, sf2, scan, scanInfo, MAXMVCC);
 290     verifyCountAndOrder(storeScanner, QUALSIZE * (startRowNum + 1),
 291         startRowNum + 1, false);
 292
 293     // Case 3.Test reversed scan with a specified start row and specified
 294     // qualifiers
 295     assertTrue(QUALSIZE > 2);
 296     scan.addColumn(FAMILYNAME, QUALS[0]);
 297     scan.addColumn(FAMILYNAME, QUALS[2]);
 298     storeScanner = getReversibleStoreScanner(memstore, sf1, sf2, scan, scanInfo, MAXMVCC);
 299     verifyCountAndOrder(storeScanner, 2 * (startRowNum + 1), startRowNum + 1,
 300         false);
 301
 302     // Case 4.Test reversed scan with mvcc based on case 3
 303     for (int readPoint = 0; readPoint < MAXMVCC; readPoint++) {
 304       LOG.info("Setting read point to " + readPoint);
 305       storeScanner = getReversibleStoreScanner(memstore, sf1, sf2, scan, scanInfo, readPoint);
 306       int expectedRowCount = 0;
 307       int expectedKVCount = 0;
 308       for (int i = startRowNum; i >= 0; i--) {
 309         int kvCount = 0;
 310         if (makeMVCC(i, 0) <= readPoint) {
 311           kvCount++;
 312         }
 313         if (makeMVCC(i, 2) <= readPoint) {
 314           kvCount++;
 315         }
 316         if (kvCount > 0) {
 317           expectedRowCount++;
 318           expectedKVCount += kvCount;
 319         }
 320       }
 321       verifyCountAndOrder(storeScanner, expectedKVCount, expectedRowCount,
 322           false);
 323     }
 324   }
 325
 326   @Test
 327   public void testReversibleRegionScanner() throws IOException {
 328     byte[] FAMILYNAME2 = Bytes.toBytes("testCf2");
 329     TableDescriptor htd = TableDescriptorBuilder.newBuilder(TableName.valueOf(name.getMethodName()))
 330       .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILYNAME))
 331       .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILYNAME2)).build();
 332     HRegion region = TEST_UTIL.createLocalHRegion(htd, null, null);
 333     loadDataToRegion(region, FAMILYNAME2);
 334
 335     // verify row count with forward scan
 336     Scan scan = new Scan();
 337     InternalScanner scanner = region.getScanner(scan);
 338     verifyCountAndOrder(scanner, ROWSIZE * QUALSIZE * 2, ROWSIZE, true);
 339
 340     // Case1:Full reversed scan
 341     scan.setReversed(true);
 342     scanner = region.getScanner(scan);
 343     verifyCountAndOrder(scanner, ROWSIZE * QUALSIZE * 2, ROWSIZE, false);
 344
 345     // Case2:Full reversed scan with one family
 346     scan = new Scan();
 347     scan.setReversed(true);
 348     scan.addFamily(FAMILYNAME);
 349     scanner = region.getScanner(scan);
 350     verifyCountAndOrder(scanner, ROWSIZE * QUALSIZE, ROWSIZE, false);
 351
 352     // Case3:Specify qualifiers + One family
 353     byte[][] specifiedQualifiers = { QUALS[1], QUALS[2] };
 354     for (byte[] specifiedQualifier : specifiedQualifiers)
 355       scan.addColumn(FAMILYNAME, specifiedQualifier);
 356     scanner = region.getScanner(scan);
 357     verifyCountAndOrder(scanner, ROWSIZE * 2, ROWSIZE, false);
 358
 359     // Case4:Specify qualifiers + Two families
 360     for (byte[] specifiedQualifier : specifiedQualifiers)
 361       scan.addColumn(FAMILYNAME2, specifiedQualifier);
 362     scanner = region.getScanner(scan);
 363     verifyCountAndOrder(scanner, ROWSIZE * 2 * 2, ROWSIZE, false);
 364
 365     // Case5: Case4 + specify start row
 366     int startRowNum = ROWSIZE * 3 / 4;
 367     scan.withStartRow(ROWS[startRowNum]);
 368     scanner = region.getScanner(scan);
 369     verifyCountAndOrder(scanner, (startRowNum + 1) * 2 * 2, (startRowNum + 1),
 370         false);
 371
 372     // Case6: Case4 + specify stop row
 373     int stopRowNum = ROWSIZE / 4;
 374     scan.withStartRow(HConstants.EMPTY_BYTE_ARRAY);
 375     scan.withStopRow(ROWS[stopRowNum]);
 376     scanner = region.getScanner(scan);
 377     verifyCountAndOrder(scanner, (ROWSIZE - stopRowNum - 1) * 2 * 2, (ROWSIZE
 378         - stopRowNum - 1), false);
 379
 380     // Case7: Case4 + specify start row + specify stop row
 381     scan.withStartRow(ROWS[startRowNum]);
 382     scanner = region.getScanner(scan);
 383     verifyCountAndOrder(scanner, (startRowNum - stopRowNum) * 2 * 2,
 384         (startRowNum - stopRowNum), false);
 385
 386     // Case8: Case7 + SingleColumnValueFilter
 387     int valueNum = startRowNum % VALUESIZE;
 388     Filter filter = new SingleColumnValueFilter(FAMILYNAME,
 389         specifiedQualifiers[0], CompareOperator.EQUAL, VALUES[valueNum]);
 390     scan.setFilter(filter);
 391     scanner = region.getScanner(scan);
 392     int unfilteredRowNum = (startRowNum - stopRowNum) / VALUESIZE
 393         + (stopRowNum / VALUESIZE == valueNum ? 0 : 1);
 394     verifyCountAndOrder(scanner, unfilteredRowNum * 2 * 2, unfilteredRowNum,
 395         false);
 396
 397     // Case9: Case7 + PageFilter
 398     int pageSize = 10;
 399     filter = new PageFilter(pageSize);
 400     scan.setFilter(filter);
 401     scanner = region.getScanner(scan);
 402     int expectedRowNum = pageSize;
 403     verifyCountAndOrder(scanner, expectedRowNum * 2 * 2, expectedRowNum, false);
 404
 405     // Case10: Case7 + FilterList+MUST_PASS_ONE
 406     SingleColumnValueFilter scvFilter1 = new SingleColumnValueFilter(
 407         FAMILYNAME, specifiedQualifiers[0], CompareOperator.EQUAL, VALUES[0]);
 408     SingleColumnValueFilter scvFilter2 = new SingleColumnValueFilter(
 409         FAMILYNAME, specifiedQualifiers[0], CompareOperator.EQUAL, VALUES[1]);
 410     expectedRowNum = 0;
 411     for (int i = startRowNum; i > stopRowNum; i--) {
 412       if (i % VALUESIZE == 0 || i % VALUESIZE == 1) {
 413         expectedRowNum++;
 414       }
 415     }
 416     filter = new FilterList(Operator.MUST_PASS_ONE, scvFilter1, scvFilter2);
 417     scan.setFilter(filter);
 418     scanner = region.getScanner(scan);
 419     verifyCountAndOrder(scanner, expectedRowNum * 2 * 2, expectedRowNum, false);
 420
 421     // Case10: Case7 + FilterList+MUST_PASS_ALL
 422     filter = new FilterList(Operator.MUST_PASS_ALL, scvFilter1, scvFilter2);
 423     expectedRowNum = 0;
 424     scan.setFilter(filter);
 425     scanner = region.getScanner(scan);
 426     verifyCountAndOrder(scanner, expectedRowNum * 2 * 2, expectedRowNum, false);
 427   }
 428
 429   private StoreScanner getReversibleStoreScanner(MemStore memstore, HStoreFile sf1, HStoreFile sf2,
 430       Scan scan, ScanInfo scanInfo, int readPoint) throws IOException {
 431     List<KeyValueScanner> scanners = getScanners(memstore, sf1, sf2, null, false, readPoint);
 432     NavigableSet<byte[]> columns = null;
 433     for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap().entrySet()) {
 434       // Should only one family
 435       columns = entry.getValue();
 436     }
 437     StoreScanner storeScanner = new ReversedStoreScanner(scan, scanInfo, columns, scanners);
 438     return storeScanner;
 439   }
 440
 441   private void verifyCountAndOrder(InternalScanner scanner,
 442       int expectedKVCount, int expectedRowCount, boolean forward)
 443       throws IOException {
 444     List<Cell> kvList = new ArrayList<>();
 445     Result lastResult = null;
 446     int rowCount = 0;
 447     int kvCount = 0;
 448     try {
 449       while (scanner.next(kvList)) {
 450         if (kvList.isEmpty()) continue;
 451         rowCount++;
 452         kvCount += kvList.size();
 453         if (lastResult != null) {
 454           Result curResult = Result.create(kvList);
 455           assertEquals("LastResult:" + lastResult + "CurResult:" + curResult,
 456               forward,
 457               Bytes.compareTo(curResult.getRow(), lastResult.getRow()) > 0);
 458         }
 459         lastResult = Result.create(kvList);
 460         kvList.clear();
 461       }
 462     } finally {
 463       scanner.close();
 464     }
 465     if (!kvList.isEmpty()) {
 466       rowCount++;
 467       kvCount += kvList.size();
 468       kvList.clear();
 469     }
 470     assertEquals(expectedKVCount, kvCount);
 471     assertEquals(expectedRowCount, rowCount);
 472   }
 473
 474   private void internalTestSeekAndNextForReversibleKeyValueHeap(
 475       ReversedKeyValueHeap kvHeap, int startRowNum) throws IOException {
 476     // Test next and seek
 477     for (int i = startRowNum; i >= 0; i--) {
 478       if (i % 2 == 1 && i - 2 >= 0) {
 479         i = i - 2;
 480         kvHeap.seekToPreviousRow(KeyValueUtil.createFirstOnRow(ROWS[i + 1]));
 481       }
 482       for (int j = 0; j < QUALSIZE; j++) {
 483         if (j % 2 == 1 && (j + 1) < QUALSIZE) {
 484           j = j + 1;
 485           kvHeap.backwardSeek(makeKV(i, j));
 486         }
 487         assertEquals(makeKV(i, j), kvHeap.peek());
 488         kvHeap.next();
 489       }
 490     }
 491     assertEquals(null, kvHeap.peek());
 492   }
 493
 494   private ReversedKeyValueHeap getReversibleKeyValueHeap(MemStore memstore, HStoreFile sf1,
 495       HStoreFile sf2, byte[] startRow, int readPoint) throws IOException {
 496     List<KeyValueScanner> scanners = getScanners(memstore, sf1, sf2, startRow, true, readPoint);
 497     ReversedKeyValueHeap kvHeap = new ReversedKeyValueHeap(scanners, CellComparatorImpl.COMPARATOR);
 498     return kvHeap;
 499   }
 500
 501   private List<KeyValueScanner> getScanners(MemStore memstore, HStoreFile sf1, HStoreFile sf2,
 502       byte[] startRow, boolean doSeek, int readPoint) throws IOException {
 503     List<StoreFileScanner> fileScanners = StoreFileScanner.getScannersForStoreFiles(
 504       Lists.newArrayList(sf1, sf2), false, true, false, false, readPoint);
 505     List<KeyValueScanner> memScanners = memstore.getScanners(readPoint);
 506     List<KeyValueScanner> scanners = new ArrayList<>(fileScanners.size() + 1);
 507     scanners.addAll(fileScanners);
 508     scanners.addAll(memScanners);
 509
 510     if (doSeek) {
 511       if (Bytes.equals(HConstants.EMPTY_START_ROW, startRow)) {
 512         for (KeyValueScanner scanner : scanners) {
 513           scanner.seekToLastRow();
 514         }
 515       } else {
 516         KeyValue startKey = KeyValueUtil.createFirstOnRow(startRow);
 517         for (KeyValueScanner scanner : scanners) {
 518           scanner.backwardSeek(startKey);
 519         }
 520       }
 521     }
 522     return scanners;
 523   }
 524
 525   private void seekTestOfReversibleKeyValueScanner(KeyValueScanner scanner)
 526       throws IOException {
 527     /**
 528      * Test without MVCC
 529      */
 530     // Test seek to last row
 531     assertTrue(scanner.seekToLastRow());
 532     assertEquals(makeKV(ROWSIZE - 1, 0), scanner.peek());
 533
 534     // Test backward seek in three cases
 535     // Case1: seek in the same row in backwardSeek
 536     KeyValue seekKey = makeKV(ROWSIZE - 2, QUALSIZE - 2);
 537     assertTrue(scanner.backwardSeek(seekKey));
 538     assertEquals(seekKey, scanner.peek());
 539
 540     // Case2: seek to the previous row in backwardSeek
 541     int seekRowNum = ROWSIZE - 2;
 542     assertTrue(scanner.backwardSeek(KeyValueUtil.createLastOnRow(ROWS[seekRowNum])));
 543     KeyValue expectedKey = makeKV(seekRowNum - 1, 0);
 544     assertEquals(expectedKey, scanner.peek());
 545
 546     // Case3: unable to backward seek
 547     assertFalse(scanner.backwardSeek(KeyValueUtil.createLastOnRow(ROWS[0])));
 548     assertEquals(null, scanner.peek());
 549
 550     // Test seek to previous row
 551     seekRowNum = ROWSIZE - 4;
 552     assertTrue(scanner.seekToPreviousRow(KeyValueUtil
 553         .createFirstOnRow(ROWS[seekRowNum])));
 554     expectedKey = makeKV(seekRowNum - 1, 0);
 555     assertEquals(expectedKey, scanner.peek());
 556
 557     // Test seek to previous row for the first row
 558     assertFalse(scanner.seekToPreviousRow(makeKV(0, 0)));
 559     assertEquals(null, scanner.peek());
 560
 561   }
 562
 563   private void seekTestOfReversibleKeyValueScannerWithMVCC(
 564       List<? extends KeyValueScanner> scanners, int readPoint) throws IOException {
 565   /**
 566    * Test with MVCC
 567    */
 568     // Test seek to last row
 569     KeyValue expectedKey = getNextReadableKeyValueWithBackwardScan(
 570         ROWSIZE - 1, 0, readPoint);
 571     boolean res = false;
 572     for (KeyValueScanner scanner : scanners) {
 573       res |= scanner.seekToLastRow();
 574     }
 575     assertEquals(expectedKey != null, res);
 576     res = false;
 577     for (KeyValueScanner scanner : scanners) {
 578       res |= (expectedKey.equals(scanner.peek()));
 579     }
 580     assertTrue(res);
 581
 582       // Test backward seek in two cases
 583       // Case1: seek in the same row in backwardSeek
 584       expectedKey = getNextReadableKeyValueWithBackwardScan(ROWSIZE - 2,
 585           QUALSIZE - 2, readPoint);
 586     res = false;
 587     for (KeyValueScanner scanner : scanners) {
 588       res |= scanner.backwardSeek(expectedKey);
 589     }
 590     assertEquals(expectedKey != null, res);
 591     res = false;
 592     for (KeyValueScanner scanner : scanners) {
 593       res |= (expectedKey.equals(scanner.peek()));
 594     }
 595     assertTrue(res);
 596
 597       // Case2: seek to the previous row in backwardSeek
 598     int seekRowNum = ROWSIZE - 3;
 599     res = false;
 600     for (KeyValueScanner scanner : scanners) {
 601       res |= scanner.backwardSeek(expectedKey);
 602     }
 603     res = false;
 604     for (KeyValueScanner scanner : scanners) {
 605       res |= (expectedKey.equals(scanner.peek()));
 606     }
 607     assertTrue(res);
 608
 609       // Test seek to previous row
 610       seekRowNum = ROWSIZE - 4;
 611       expectedKey = getNextReadableKeyValueWithBackwardScan(seekRowNum - 1, 0,
 612           readPoint);
 613     res = false;
 614     for (KeyValueScanner scanner : scanners) {
 615       res |= scanner.seekToPreviousRow(KeyValueUtil.createFirstOnRow(ROWS[seekRowNum]));
 616     }
 617     assertEquals(expectedKey != null, res);
 618     res = false;
 619     for (KeyValueScanner scanner : scanners) {
 620       res |= (expectedKey.equals(scanner.peek()));
 621     }
 622     assertTrue(res);
 623   }
 624
 625   private KeyValue getNextReadableKeyValueWithBackwardScan(int startRowNum,
 626       int startQualNum, int readPoint) {
 627     Pair<Integer, Integer> nextReadableNum = getNextReadableNumWithBackwardScan(
 628         startRowNum, startQualNum, readPoint);
 629     if (nextReadableNum == null)
 630       return null;
 631     return makeKV(nextReadableNum.getFirst(), nextReadableNum.getSecond());
 632   }
 633
 634   private Pair<Integer, Integer> getNextReadableNumWithBackwardScan(
 635       int startRowNum, int startQualNum, int readPoint) {
 636     Pair<Integer, Integer> nextReadableNum = null;
 637     boolean findExpected = false;
 638     for (int i = startRowNum; i >= 0; i--) {
 639       for (int j = (i == startRowNum ? startQualNum : 0); j < QUALSIZE; j++) {
 640         if (makeMVCC(i, j) <= readPoint) {
 641           nextReadableNum = new Pair<>(i, j);
 642           findExpected = true;
 643           break;
 644         }
 645       }
 646       if (findExpected)
 647         break;
 648     }
 649     return nextReadableNum;
 650   }
 651
 652   private static void loadDataToRegion(HRegion region, byte[] additionalFamily)
 653       throws IOException {
 654     for (int i = 0; i < ROWSIZE; i++) {
 655       Put put = new Put(ROWS[i]);
 656       for (int j = 0; j < QUALSIZE; j++) {
 657         put.add(makeKV(i, j));
 658         // put additional family
 659         put.add(makeKV(i, j, additionalFamily));
 660       }
 661       region.put(put);
 662       if (i == ROWSIZE / 3 || i == ROWSIZE * 2 / 3) {
 663         region.flush(true);
 664       }
 665     }
 666   }
 667
 668   private static void writeMemstoreAndStoreFiles(MemStore memstore,
 669       final StoreFileWriter[] writers) throws IOException {
 670     try {
 671       for (int i = 0; i < ROWSIZE; i++) {
 672         for (int j = 0; j < QUALSIZE; j++) {
 673           if (i % 2 == 0) {
 674             memstore.add(makeKV(i, j), null);
 675           } else {
 676             writers[(i + j) % writers.length].append(makeKV(i, j));
 677           }
 678         }
 679       }
 680     } finally {
 681       for (int i = 0; i < writers.length; i++) {
 682         writers[i].close();
 683       }
 684     }
 685   }
 686
 687   private static void writeStoreFile(final StoreFileWriter writer)
 688       throws IOException {
 689     try {
 690       for (int i = 0; i < ROWSIZE; i++) {
 691         for (int j = 0; j < QUALSIZE; j++) {
 692           writer.append(makeKV(i, j));
 693         }
 694       }
 695     } finally {
 696       writer.close();
 697     }
 698   }
 699
 700   private static void writeMemstore(MemStore memstore) throws IOException {
 701     // Add half of the keyvalues to memstore
 702     for (int i = 0; i < ROWSIZE; i++) {
 703       for (int j = 0; j < QUALSIZE; j++) {
 704         if ((i + j) % 2 == 0) {
 705           memstore.add(makeKV(i, j), null);
 706         }
 707       }
 708     }
 709     memstore.snapshot();
 710     // Add another half of the keyvalues to snapshot
 711     for (int i = 0; i < ROWSIZE; i++) {
 712       for (int j = 0; j < QUALSIZE; j++) {
 713         if ((i + j) % 2 == 1) {
 714           memstore.add(makeKV(i, j), null);
 715         }
 716       }
 717     }
 718   }
 719
 720   private static KeyValue makeKV(int rowNum, int cqNum) {
 721     return makeKV(rowNum, cqNum, FAMILYNAME);
 722   }
 723
 724   private static KeyValue makeKV(int rowNum, int cqNum, byte[] familyName) {
 725     KeyValue kv = new KeyValue(ROWS[rowNum], familyName, QUALS[cqNum], TS,
 726         VALUES[rowNum % VALUESIZE]);
 727     kv.setSequenceId(makeMVCC(rowNum, cqNum));
 728     return kv;
 729   }
 730
 731   private static long makeMVCC(int rowNum, int cqNum) {
 732     return (rowNum + cqNum) % (MAXMVCC + 1);
 733   }
 734
 735   private static byte[][] makeN(byte[] base, int n) {
 736     byte[][] ret = new byte[n][];
 737     for (int i = 0; i < n; i++) {
 738       ret[i] = Bytes.add(base, Bytes.toBytes(String.format("%04d", i)));
 739     }
 740     return ret;
 741   }
 742 }