001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.client;
019
020import static org.junit.Assert.assertArrayEquals;
021import static org.junit.Assert.assertEquals;
022import static org.junit.Assert.assertFalse;
023import static org.junit.Assert.assertNull;
024import static org.junit.Assert.assertTrue;
025import static org.junit.Assert.fail;
026
027import java.io.IOException;
028import java.util.ArrayList;
029import java.util.List;
030import java.util.concurrent.TimeUnit;
031import java.util.function.Consumer;
032import java.util.stream.IntStream;
033import org.apache.hadoop.conf.Configuration;
034import org.apache.hadoop.hbase.Cell;
035import org.apache.hadoop.hbase.CompareOperator;
036import org.apache.hadoop.hbase.HBaseClassTestRule;
037import org.apache.hadoop.hbase.HBaseTestingUtility;
038import org.apache.hadoop.hbase.HColumnDescriptor;
039import org.apache.hadoop.hbase.HConstants;
040import org.apache.hadoop.hbase.HRegionInfo;
041import org.apache.hadoop.hbase.HRegionLocation;
042import org.apache.hadoop.hbase.HTestConst;
043import org.apache.hadoop.hbase.KeyValue;
044import org.apache.hadoop.hbase.MiniHBaseCluster;
045import org.apache.hadoop.hbase.TableName;
046import org.apache.hadoop.hbase.TableNotFoundException;
047import org.apache.hadoop.hbase.filter.BinaryComparator;
048import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
049import org.apache.hadoop.hbase.filter.ColumnRangeFilter;
050import org.apache.hadoop.hbase.filter.QualifierFilter;
051import org.apache.hadoop.hbase.regionserver.HRegionServer;
052import org.apache.hadoop.hbase.testclassification.ClientTests;
053import org.apache.hadoop.hbase.testclassification.MediumTests;
054import org.apache.hadoop.hbase.util.Bytes;
055import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
056import org.junit.After;
057import org.junit.AfterClass;
058import org.junit.Before;
059import org.junit.BeforeClass;
060import org.junit.ClassRule;
061import org.junit.Rule;
062import org.junit.Test;
063import org.junit.experimental.categories.Category;
064import org.junit.rules.TestName;
065import org.slf4j.Logger;
066import org.slf4j.LoggerFactory;
067
068/**
069 * A client-side test, mostly testing scanners with various parameters.
070 */
071@Category({MediumTests.class, ClientTests.class})
072public class TestScannersFromClientSide {
073
074  @ClassRule
075  public static final HBaseClassTestRule CLASS_RULE =
076      HBaseClassTestRule.forClass(TestScannersFromClientSide.class);
077
078  private static final Logger LOG = LoggerFactory.getLogger(TestScannersFromClientSide.class);
079
080  private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
081  private static byte [] ROW = Bytes.toBytes("testRow");
082  private static byte [] FAMILY = Bytes.toBytes("testFamily");
083  private static byte [] QUALIFIER = Bytes.toBytes("testQualifier");
084  private static byte [] VALUE = Bytes.toBytes("testValue");
085
086  @Rule
087  public TestName name = new TestName();
088
089  /**
090   * @throws java.lang.Exception
091   */
092  @BeforeClass
093  public static void setUpBeforeClass() throws Exception {
094    Configuration conf = TEST_UTIL.getConfiguration();
095    conf.setLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, 10 * 1024 * 1024);
096    TEST_UTIL.startMiniCluster(3);
097  }
098
099  /**
100   * @throws java.lang.Exception
101   */
102  @AfterClass
103  public static void tearDownAfterClass() throws Exception {
104    TEST_UTIL.shutdownMiniCluster();
105  }
106
107  /**
108   * @throws java.lang.Exception
109   */
110  @Before
111  public void setUp() throws Exception {
112    // Nothing to do.
113  }
114
115  /**
116   * @throws java.lang.Exception
117   */
118  @After
119  public void tearDown() throws Exception {
120    // Nothing to do.
121  }
122
123  /**
124   * Test from client side for batch of scan
125   *
126   * @throws Exception
127   */
128  @Test
129  public void testScanBatch() throws Exception {
130    final TableName tableName = TableName.valueOf(name.getMethodName());
131    byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 8);
132
133    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
134
135    Put put;
136    Scan scan;
137    Delete delete;
138    Result result;
139    ResultScanner scanner;
140    boolean toLog = true;
141    List<Cell> kvListExp;
142
143    // table: row, family, c0:0, c1:1, ... , c7:7
144    put = new Put(ROW);
145    for (int i=0; i < QUALIFIERS.length; i++) {
146      KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
147      put.add(kv);
148    }
149    ht.put(put);
150
151    // table: row, family, c0:0, c1:1, ..., c6:2, c6:6 , c7:7
152    put = new Put(ROW);
153    KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[6], 2, VALUE);
154    put.add(kv);
155    ht.put(put);
156
157    // delete upto ts: 3
158    delete = new Delete(ROW);
159    delete.addFamily(FAMILY, 3);
160    ht.delete(delete);
161
162    // without batch
163    scan = new Scan().withStartRow(ROW);
164    scan.setMaxVersions();
165    scanner = ht.getScanner(scan);
166
167    // c4:4, c5:5, c6:6, c7:7
168    kvListExp = new ArrayList<>();
169    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
170    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
171    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
172    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
173    result = scanner.next();
174    verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
175
176    // with batch
177    scan =  new Scan().withStartRow(ROW);
178    scan.setMaxVersions();
179    scan.setBatch(2);
180    scanner = ht.getScanner(scan);
181
182    // First batch: c4:4, c5:5
183    kvListExp = new ArrayList<>();
184    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
185    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
186    result = scanner.next();
187    verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
188
189    // Second batch: c6:6, c7:7
190    kvListExp = new ArrayList<>();
191    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
192    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
193    result = scanner.next();
194    verifyResult(result, kvListExp, toLog, "Testing second batch of scan");
195
196  }
197
198  @Test
199  public void testMaxResultSizeIsSetToDefault() throws Exception {
200    final TableName tableName = TableName.valueOf(name.getMethodName());
201    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
202
203    // The max result size we expect the scan to use by default.
204    long expectedMaxResultSize =
205        TEST_UTIL.getConfiguration().getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
206          HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
207
208    int numRows = 5;
209    byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
210
211    int numQualifiers = 10;
212    byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
213
214    // Specify the cell size such that a single row will be larger than the default
215    // value of maxResultSize. This means that Scan RPCs should return at most a single
216    // result back to the client.
217    int cellSize = (int) (expectedMaxResultSize / (numQualifiers - 1));
218    byte[] cellValue = Bytes.createMaxByteArray(cellSize);
219
220    Put put;
221    List<Put> puts = new ArrayList<>();
222    for (int row = 0; row < ROWS.length; row++) {
223      put = new Put(ROWS[row]);
224      for (int qual = 0; qual < QUALIFIERS.length; qual++) {
225        KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], cellValue);
226        put.add(kv);
227      }
228      puts.add(put);
229    }
230    ht.put(puts);
231
232    // Create a scan with the default configuration.
233    Scan scan = new Scan();
234
235    ResultScanner scanner = ht.getScanner(scan);
236    assertTrue(scanner instanceof ClientScanner);
237    ClientScanner clientScanner = (ClientScanner) scanner;
238
239    // Call next to issue a single RPC to the server
240    scanner.next();
241
242    // The scanner should have, at most, a single result in its cache. If there more results exists
243    // in the cache it means that more than the expected max result size was fetched.
244    assertTrue("The cache contains: " + clientScanner.getCacheSize() + " results",
245      clientScanner.getCacheSize() <= 1);
246  }
247
248  /**
249   * Scan on not existing table should throw the exception with correct message
250   */
251  @Test
252  public void testScannerForNotExistingTable() {
253    String[] tableNames = {"A", "Z", "A:A", "Z:Z"};
254    for(String tableName : tableNames) {
255      try {
256        Table table = TEST_UTIL.getConnection().getTable(TableName.valueOf(tableName));
257        testSmallScan(table, true, 1, 5);
258        fail("TableNotFoundException was not thrown");
259      } catch (TableNotFoundException e) {
260        // We expect that the message for TableNotFoundException would have only the table name only
261        // Otherwise that would mean that localeRegionInMeta doesn't work properly
262        assertEquals(e.getMessage(), tableName);
263      } catch (Exception e) {
264        fail("Unexpected exception " + e.getMessage());
265      }
266    }
267  }
268
269  @Test
270  public void testSmallScan() throws Exception {
271    final TableName tableName = TableName.valueOf(name.getMethodName());
272
273    int numRows = 10;
274    byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
275
276    int numQualifiers = 10;
277    byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
278
279    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
280
281    Put put;
282    List<Put> puts = new ArrayList<>();
283    for (int row = 0; row < ROWS.length; row++) {
284      put = new Put(ROWS[row]);
285      for (int qual = 0; qual < QUALIFIERS.length; qual++) {
286        KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], VALUE);
287        put.add(kv);
288      }
289      puts.add(put);
290    }
291    ht.put(puts);
292
293    int expectedRows = numRows;
294    int expectedCols = numRows * numQualifiers;
295
296    // Test normal and reversed
297    testSmallScan(ht, true, expectedRows, expectedCols);
298    testSmallScan(ht, false, expectedRows, expectedCols);
299  }
300
301  /**
302   * Run through a variety of test configurations with a small scan
303   * @param table
304   * @param reversed
305   * @param rows
306   * @param columns
307   * @throws Exception
308   */
309  private void testSmallScan(Table table, boolean reversed, int rows, int columns) throws Exception {
310    Scan baseScan = new Scan();
311    baseScan.setReversed(reversed);
312    baseScan.setSmall(true);
313
314    Scan scan = new Scan(baseScan);
315    verifyExpectedCounts(table, scan, rows, columns);
316
317    scan = new Scan(baseScan);
318    scan.setMaxResultSize(1);
319    verifyExpectedCounts(table, scan, rows, columns);
320
321    scan = new Scan(baseScan);
322    scan.setMaxResultSize(1);
323    scan.setCaching(Integer.MAX_VALUE);
324    verifyExpectedCounts(table, scan, rows, columns);
325  }
326
327  private void verifyExpectedCounts(Table table, Scan scan, int expectedRowCount,
328      int expectedCellCount) throws Exception {
329    ResultScanner scanner = table.getScanner(scan);
330
331    int rowCount = 0;
332    int cellCount = 0;
333    Result r = null;
334    while ((r = scanner.next()) != null) {
335      rowCount++;
336      cellCount += r.rawCells().length;
337    }
338
339    assertTrue("Expected row count: " + expectedRowCount + " Actual row count: " + rowCount,
340        expectedRowCount == rowCount);
341    assertTrue("Expected cell count: " + expectedCellCount + " Actual cell count: " + cellCount,
342        expectedCellCount == cellCount);
343    scanner.close();
344  }
345
346  /**
347   * Test from client side for get with maxResultPerCF set
348   *
349   * @throws Exception
350   */
351  @Test
352  public void testGetMaxResults() throws Exception {
353    final TableName tableName = TableName.valueOf(name.getMethodName());
354    byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
355    byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
356
357    Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
358
359    Get get;
360    Put put;
361    Result result;
362    boolean toLog = true;
363    List<Cell> kvListExp;
364
365    kvListExp = new ArrayList<>();
366    // Insert one CF for row[0]
367    put = new Put(ROW);
368    for (int i=0; i < 10; i++) {
369      KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
370      put.add(kv);
371      kvListExp.add(kv);
372    }
373    ht.put(put);
374
375    get = new Get(ROW);
376    result = ht.get(get);
377    verifyResult(result, kvListExp, toLog, "Testing without setting maxResults");
378
379    get = new Get(ROW);
380    get.setMaxResultsPerColumnFamily(2);
381    result = ht.get(get);
382    kvListExp = new ArrayList<>();
383    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[0], 1, VALUE));
384    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
385    verifyResult(result, kvListExp, toLog, "Testing basic setMaxResults");
386
387    // Filters: ColumnRangeFilter
388    get = new Get(ROW);
389    get.setMaxResultsPerColumnFamily(5);
390    get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5],
391                                        true));
392    result = ht.get(get);
393    kvListExp = new ArrayList<>();
394    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[2], 1, VALUE));
395    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
396    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
397    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
398    verifyResult(result, kvListExp, toLog, "Testing single CF with CRF");
399
400    // Insert two more CF for row[0]
401    // 20 columns for CF2, 10 columns for CF1
402    put = new Put(ROW);
403    for (int i=0; i < QUALIFIERS.length; i++) {
404      KeyValue kv = new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE);
405      put.add(kv);
406    }
407    ht.put(put);
408
409    put = new Put(ROW);
410    for (int i=0; i < 10; i++) {
411      KeyValue kv = new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE);
412      put.add(kv);
413    }
414    ht.put(put);
415
416    get = new Get(ROW);
417    get.setMaxResultsPerColumnFamily(12);
418    get.addFamily(FAMILIES[1]);
419    get.addFamily(FAMILIES[2]);
420    result = ht.get(get);
421    kvListExp = new ArrayList<>();
422    //Exp: CF1:q0, ..., q9, CF2: q0, q1, q10, q11, ..., q19
423    for (int i=0; i < 10; i++) {
424      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
425    }
426    for (int i=0; i < 2; i++) {
427        kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
428      }
429    for (int i=10; i < 20; i++) {
430      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
431    }
432    verifyResult(result, kvListExp, toLog, "Testing multiple CFs");
433
434    // Filters: ColumnRangeFilter and ColumnPrefixFilter
435    get = new Get(ROW);
436    get.setMaxResultsPerColumnFamily(3);
437    get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, null, true));
438    result = ht.get(get);
439    kvListExp = new ArrayList<>();
440    for (int i=2; i < 5; i++) {
441      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
442    }
443    for (int i=2; i < 5; i++) {
444      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
445    }
446    for (int i=2; i < 5; i++) {
447      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
448    }
449    verifyResult(result, kvListExp, toLog, "Testing multiple CFs + CRF");
450
451    get = new Get(ROW);
452    get.setMaxResultsPerColumnFamily(7);
453    get.setFilter(new ColumnPrefixFilter(QUALIFIERS[1]));
454    result = ht.get(get);
455    kvListExp = new ArrayList<>();
456    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
457    kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[1], 1, VALUE));
458    kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[1], 1, VALUE));
459    for (int i=10; i < 16; i++) {
460      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
461    }
462    verifyResult(result, kvListExp, toLog, "Testing multiple CFs + PFF");
463
464  }
465
466  /**
467   * Test from client side for scan with maxResultPerCF set
468   *
469   * @throws Exception
470   */
471  @Test
472  public void testScanMaxResults() throws Exception {
473    final TableName tableName = TableName.valueOf(name.getMethodName());
474    byte [][] ROWS = HTestConst.makeNAscii(ROW, 2);
475    byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
476    byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 10);
477
478    Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
479
480    Put put;
481    Scan scan;
482    Result result;
483    boolean toLog = true;
484    List<Cell> kvListExp, kvListScan;
485
486    kvListExp = new ArrayList<>();
487
488    for (int r=0; r < ROWS.length; r++) {
489      put = new Put(ROWS[r]);
490      for (int c=0; c < FAMILIES.length; c++) {
491        for (int q=0; q < QUALIFIERS.length; q++) {
492          KeyValue kv = new KeyValue(ROWS[r], FAMILIES[c], QUALIFIERS[q], 1, VALUE);
493          put.add(kv);
494          if (q < 4) {
495            kvListExp.add(kv);
496          }
497        }
498      }
499      ht.put(put);
500    }
501
502    scan = new Scan();
503    scan.setMaxResultsPerColumnFamily(4);
504    ResultScanner scanner = ht.getScanner(scan);
505    kvListScan = new ArrayList<>();
506    while ((result = scanner.next()) != null) {
507      for (Cell kv : result.listCells()) {
508        kvListScan.add(kv);
509      }
510    }
511    result = Result.create(kvListScan);
512    verifyResult(result, kvListExp, toLog, "Testing scan with maxResults");
513
514  }
515
516  /**
517   * Test from client side for get with rowOffset
518   *
519   * @throws Exception
520   */
521  @Test
522  public void testGetRowOffset() throws Exception {
523    final TableName tableName = TableName.valueOf(name.getMethodName());
524    byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
525    byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
526
527    Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
528
529    Get get;
530    Put put;
531    Result result;
532    boolean toLog = true;
533    List<Cell> kvListExp;
534
535    // Insert one CF for row
536    kvListExp = new ArrayList<>();
537    put = new Put(ROW);
538    for (int i=0; i < 10; i++) {
539      KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
540      put.add(kv);
541      // skipping first two kvs
542      if (i < 2) continue;
543      kvListExp.add(kv);
544    }
545    ht.put(put);
546
547    //setting offset to 2
548    get = new Get(ROW);
549    get.setRowOffsetPerColumnFamily(2);
550    result = ht.get(get);
551    verifyResult(result, kvListExp, toLog, "Testing basic setRowOffset");
552
553    //setting offset to 20
554    get = new Get(ROW);
555    get.setRowOffsetPerColumnFamily(20);
556    result = ht.get(get);
557    kvListExp = new ArrayList<>();
558    verifyResult(result, kvListExp, toLog, "Testing offset > #kvs");
559
560    //offset + maxResultPerCF
561    get = new Get(ROW);
562    get.setRowOffsetPerColumnFamily(4);
563    get.setMaxResultsPerColumnFamily(5);
564    result = ht.get(get);
565    kvListExp = new ArrayList<>();
566    for (int i=4; i < 9; i++) {
567      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
568    }
569    verifyResult(result, kvListExp, toLog,
570      "Testing offset + setMaxResultsPerCF");
571
572    // Filters: ColumnRangeFilter
573    get = new Get(ROW);
574    get.setRowOffsetPerColumnFamily(1);
575    get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5],
576                                        true));
577    result = ht.get(get);
578    kvListExp = new ArrayList<>();
579    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
580    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
581    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
582    verifyResult(result, kvListExp, toLog, "Testing offset with CRF");
583
584    // Insert into two more CFs for row
585    // 10 columns for CF2, 10 columns for CF1
586    for(int j=2; j > 0; j--) {
587      put = new Put(ROW);
588      for (int i=0; i < 10; i++) {
589        KeyValue kv = new KeyValue(ROW, FAMILIES[j], QUALIFIERS[i], 1, VALUE);
590        put.add(kv);
591      }
592      ht.put(put);
593    }
594
595    get = new Get(ROW);
596    get.setRowOffsetPerColumnFamily(4);
597    get.setMaxResultsPerColumnFamily(2);
598    get.addFamily(FAMILIES[1]);
599    get.addFamily(FAMILIES[2]);
600    result = ht.get(get);
601    kvListExp = new ArrayList<>();
602    //Exp: CF1:q4, q5, CF2: q4, q5
603    kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[4], 1, VALUE));
604    kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[5], 1, VALUE));
605    kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[4], 1, VALUE));
606    kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[5], 1, VALUE));
607    verifyResult(result, kvListExp, toLog,
608       "Testing offset + multiple CFs + maxResults");
609  }
610
611  /**
612   * Test from client side for scan while the region is reopened
613   * on the same region server.
614   *
615   * @throws Exception
616   */
617  @Test
618  public void testScanOnReopenedRegion() throws Exception {
619    final TableName tableName = TableName.valueOf(name.getMethodName());
620    byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 2);
621
622    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
623
624    Put put;
625    Scan scan;
626    Result result;
627    ResultScanner scanner;
628    boolean toLog = false;
629    List<Cell> kvListExp;
630
631    // table: row, family, c0:0, c1:1
632    put = new Put(ROW);
633    for (int i=0; i < QUALIFIERS.length; i++) {
634      KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
635      put.add(kv);
636    }
637    ht.put(put);
638
639    scan = new Scan().withStartRow(ROW);
640    scanner = ht.getScanner(scan);
641
642    HRegionLocation loc;
643
644    try (RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName)) {
645      loc = locator.getRegionLocation(ROW);
646    }
647    HRegionInfo hri = loc.getRegionInfo();
648    MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
649    byte[] regionName = hri.getRegionName();
650    int i = cluster.getServerWith(regionName);
651    HRegionServer rs = cluster.getRegionServer(i);
652    LOG.info("Unassigning " + hri);
653    TEST_UTIL.getAdmin().unassign(hri.getRegionName(), true);
654    long startTime = EnvironmentEdgeManager.currentTime();
655    long timeOut = 10000;
656    boolean offline = false;
657    while (true) {
658      if (rs.getOnlineRegion(regionName) == null) {
659        offline = true;
660        break;
661      }
662      assertTrue("Timed out in closing the testing region",
663        EnvironmentEdgeManager.currentTime() < startTime + timeOut);
664    }
665    assertTrue(offline);
666    LOG.info("Assigning " + hri);
667    TEST_UTIL.getAdmin().assign(hri.getRegionName());
668    startTime = EnvironmentEdgeManager.currentTime();
669    while (true) {
670      rs = cluster.getRegionServer(cluster.getServerWith(regionName));
671      if (rs != null && rs.getOnlineRegion(regionName) != null) {
672        offline = false;
673        break;
674      }
675      assertTrue("Timed out in open the testing region",
676        EnvironmentEdgeManager.currentTime() < startTime + timeOut);
677    }
678    assertFalse(offline);
679
680    // c0:0, c1:1
681    kvListExp = new ArrayList<>();
682    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[0], 0, VALUE));
683    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[1], 1, VALUE));
684    result = scanner.next();
685    verifyResult(result, kvListExp, toLog, "Testing scan on re-opened region");
686  }
687
688  @Test
689  public void testAsyncScannerWithSmallData() throws Exception {
690    testAsyncScanner(TableName.valueOf(name.getMethodName()),
691      2,
692      3,
693      10,
694      -1,
695      null);
696  }
697
698  @Test
699  public void testAsyncScannerWithManyRows() throws Exception {
700    testAsyncScanner(TableName.valueOf(name.getMethodName()),
701      30000,
702      1,
703      1,
704      -1,
705      null);
706  }
707
708  @Test
709  public void testAsyncScannerWithoutCaching() throws Exception {
710    testAsyncScanner(TableName.valueOf(name.getMethodName()),
711      5,
712      1,
713      1,
714      1,
715      (b) -> {
716        try {
717          TimeUnit.MILLISECONDS.sleep(500);
718        } catch (InterruptedException ex) {
719        }
720      });
721  }
722
723  private void testAsyncScanner(TableName table, int rowNumber, int familyNumber,
724      int qualifierNumber, int caching, Consumer<Boolean> listener) throws Exception {
725    assert rowNumber > 0;
726    assert familyNumber > 0;
727    assert qualifierNumber > 0;
728    byte[] row = Bytes.toBytes("r");
729    byte[] family = Bytes.toBytes("f");
730    byte[] qualifier = Bytes.toBytes("q");
731    byte[][] rows = makeNAsciiWithZeroPrefix(row, rowNumber);
732    byte[][] families = makeNAsciiWithZeroPrefix(family, familyNumber);
733    byte[][] qualifiers = makeNAsciiWithZeroPrefix(qualifier, qualifierNumber);
734
735    Table ht = TEST_UTIL.createTable(table, families);
736
737    boolean toLog = true;
738    List<Cell> kvListExp = new ArrayList<>();
739
740    List<Put> puts = new ArrayList<>();
741    for (byte[] r : rows) {
742      Put put = new Put(r);
743      for (byte[] f : families) {
744        for (byte[] q : qualifiers) {
745          KeyValue kv = new KeyValue(r, f, q, 1, VALUE);
746          put.add(kv);
747          kvListExp.add(kv);
748        }
749      }
750      puts.add(put);
751      if (puts.size() > 1000) {
752        ht.put(puts);
753        puts.clear();
754      }
755    }
756    if (!puts.isEmpty()) {
757      ht.put(puts);
758      puts.clear();
759    }
760
761    Scan scan = new Scan();
762    scan.setAsyncPrefetch(true);
763    if (caching > 0) {
764      scan.setCaching(caching);
765    }
766    try (ResultScanner scanner = ht.getScanner(scan)) {
767      assertTrue("Not instance of async scanner",scanner instanceof ClientAsyncPrefetchScanner);
768      ((ClientAsyncPrefetchScanner) scanner).setPrefetchListener(listener);
769      List<Cell> kvListScan = new ArrayList<>();
770      Result result;
771      boolean first = true;
772      int actualRows = 0;
773      while ((result = scanner.next()) != null) {
774        ++actualRows;
775        // waiting for cache. see HBASE-17376
776        if (first) {
777          TimeUnit.SECONDS.sleep(1);
778          first = false;
779        }
780        for (Cell kv : result.listCells()) {
781          kvListScan.add(kv);
782        }
783      }
784      assertEquals(rowNumber, actualRows);
785      // These cells may have different rows but it is ok. The Result#getRow
786      // isn't used in the verifyResult()
787      result = Result.create(kvListScan);
788      verifyResult(result, kvListExp, toLog, "Testing async scan");
789    }
790
791    TEST_UTIL.deleteTable(table);
792  }
793
794  private static byte[][] makeNAsciiWithZeroPrefix(byte[] base, int n) {
795    int maxLength = Integer.toString(n).length();
796    byte [][] ret = new byte[n][];
797    for (int i = 0; i < n; i++) {
798      int length = Integer.toString(i).length();
799      StringBuilder buf = new StringBuilder(Integer.toString(i));
800      IntStream.range(0, maxLength - length).forEach(v -> buf.insert(0, "0"));
801      byte[] tail = Bytes.toBytes(buf.toString());
802      ret[i] = Bytes.add(base, tail);
803    }
804    return ret;
805  }
806
807  static void verifyResult(Result result, List<Cell> expKvList, boolean toLog,
808      String msg) {
809
810    LOG.info(msg);
811    LOG.info("Expected count: " + expKvList.size());
812    LOG.info("Actual count: " + result.size());
813    if (expKvList.isEmpty())
814      return;
815
816    int i = 0;
817    for (Cell kv : result.rawCells()) {
818      if (i >= expKvList.size()) {
819        break;  // we will check the size later
820      }
821
822      Cell kvExp = expKvList.get(i++);
823      if (toLog) {
824        LOG.info("get kv is: " + kv.toString());
825        LOG.info("exp kv is: " + kvExp.toString());
826      }
827      assertTrue("Not equal", kvExp.equals(kv));
828    }
829
830    assertEquals(expKvList.size(), result.size());
831  }
832
833  @Test
834  public void testReadExpiredDataForRawScan() throws IOException {
835    TableName tableName = TableName.valueOf(name.getMethodName());
836    long ts = System.currentTimeMillis() - 10000;
837    byte[] value = Bytes.toBytes("expired");
838    try (Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
839      table.put(new Put(ROW).addColumn(FAMILY, QUALIFIER, ts, value));
840      assertArrayEquals(value, table.get(new Get(ROW)).getValue(FAMILY, QUALIFIER));
841      TEST_UTIL.getAdmin().modifyColumnFamily(tableName,
842        new HColumnDescriptor(FAMILY).setTimeToLive(5));
843      try (ResultScanner scanner = table.getScanner(FAMILY)) {
844        assertNull(scanner.next());
845      }
846      try (ResultScanner scanner = table.getScanner(new Scan().setRaw(true))) {
847        assertArrayEquals(value, scanner.next().getValue(FAMILY, QUALIFIER));
848        assertNull(scanner.next());
849      }
850    }
851  }
852
853  @Test
854  public void testScanWithColumnsAndFilterAndVersion() throws IOException {
855    TableName tableName = TableName.valueOf(name.getMethodName());
856    try (Table table = TEST_UTIL.createTable(tableName, FAMILY, 4)) {
857      for (int i = 0; i < 4; i++) {
858        Put put = new Put(ROW);
859        put.addColumn(FAMILY, QUALIFIER, VALUE);
860        table.put(put);
861      }
862
863      Scan scan = new Scan();
864      scan.addColumn(FAMILY, QUALIFIER);
865      scan.setFilter(new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(QUALIFIER)));
866      scan.readVersions(3);
867
868      try (ResultScanner scanner = table.getScanner(scan)) {
869        Result result = scanner.next();
870        assertEquals(3, result.size());
871      }
872    }
873  }
874}