hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScannersFromClientSide.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.client;
  19
  20 import static org.apache.hadoop.hbase.HConstants.RPC_CODEC_CONF_KEY;
  21 import static org.apache.hadoop.hbase.client.TestFromClientSide3.generateHugeValue;
  22 import static org.apache.hadoop.hbase.ipc.RpcClient.DEFAULT_CODEC_CLASS;
  23 import static org.junit.Assert.assertArrayEquals;
  24 import static org.junit.Assert.assertEquals;
  25 import static org.junit.Assert.assertFalse;
  26 import static org.junit.Assert.assertNotNull;
  27 import static org.junit.Assert.assertNull;
  28 import static org.junit.Assert.assertTrue;
  29 import static org.junit.Assert.fail;
  30
  31 import java.io.IOException;
  32 import java.util.ArrayList;
  33 import java.util.List;
  34 import java.util.concurrent.TimeUnit;
  35 import java.util.function.Consumer;
  36 import java.util.stream.IntStream;
  37 import org.apache.hadoop.conf.Configuration;
  38 import org.apache.hadoop.hbase.Cell;
  39 import org.apache.hadoop.hbase.CompareOperator;
  40 import org.apache.hadoop.hbase.HBaseClassTestRule;
  41 import org.apache.hadoop.hbase.HBaseTestingUtility;
  42 import org.apache.hadoop.hbase.HColumnDescriptor;
  43 import org.apache.hadoop.hbase.HConstants;
  44 import org.apache.hadoop.hbase.HRegionInfo;
  45 import org.apache.hadoop.hbase.HRegionLocation;
  46 import org.apache.hadoop.hbase.HTestConst;
  47 import org.apache.hadoop.hbase.KeyValue;
  48 import org.apache.hadoop.hbase.MiniHBaseCluster;
  49 import org.apache.hadoop.hbase.TableName;
  50 import org.apache.hadoop.hbase.TableNotFoundException;
  51 import org.apache.hadoop.hbase.filter.BinaryComparator;
  52 import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
  53 import org.apache.hadoop.hbase.filter.ColumnRangeFilter;
  54 import org.apache.hadoop.hbase.filter.QualifierFilter;
  55 import org.apache.hadoop.hbase.regionserver.HRegionServer;
  56 import org.apache.hadoop.hbase.testclassification.ClientTests;
  57 import org.apache.hadoop.hbase.testclassification.MediumTests;
  58 import org.apache.hadoop.hbase.util.Bytes;
  59 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
  60 import org.junit.After;
  61 import org.junit.AfterClass;
  62 import org.junit.Before;
  63 import org.junit.BeforeClass;
  64 import org.junit.ClassRule;
  65 import org.junit.Rule;
  66 import org.junit.Test;
  67 import org.junit.experimental.categories.Category;
  68 import org.junit.rules.TestName;
  69 import org.slf4j.Logger;
  70 import org.slf4j.LoggerFactory;
  71
  72 /**
  73  * A client-side test, mostly testing scanners with various parameters.
  74  */
  75 @Category({MediumTests.class, ClientTests.class})
  76 public class TestScannersFromClientSide {
  77
  78   @ClassRule
  79   public static final HBaseClassTestRule CLASS_RULE =
  80       HBaseClassTestRule.forClass(TestScannersFromClientSide.class);
  81
  82   private static final Logger LOG = LoggerFactory.getLogger(TestScannersFromClientSide.class);
  83
  84   private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
  85   private static byte [] ROW = Bytes.toBytes("testRow");
  86   private static byte [] FAMILY = Bytes.toBytes("testFamily");
  87   private static byte [] QUALIFIER = Bytes.toBytes("testQualifier");
  88   private static byte [] VALUE = Bytes.toBytes("testValue");
  89
  90   @Rule
  91   public TestName name = new TestName();
  92
  93   /**
  94    * @throws java.lang.Exception
  95    */
  96   @BeforeClass
  97   public static void setUpBeforeClass() throws Exception {
  98     Configuration conf = TEST_UTIL.getConfiguration();
  99     conf.setLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, 10 * 1024 * 1024);
 100     TEST_UTIL.startMiniCluster(3);
 101   }
 102
 103   /**
 104    * @throws java.lang.Exception
 105    */
 106   @AfterClass
 107   public static void tearDownAfterClass() throws Exception {
 108     TEST_UTIL.shutdownMiniCluster();
 109   }
 110
 111   /**
 112    * @throws java.lang.Exception
 113    */
 114   @Before
 115   public void setUp() throws Exception {
 116     // Nothing to do.
 117   }
 118
 119   /**
 120    * @throws java.lang.Exception
 121    */
 122   @After
 123   public void tearDown() throws Exception {
 124     // Nothing to do.
 125   }
 126
 127   /**
 128    * Test from client side for batch of scan
 129    *
 130    * @throws Exception
 131    */
 132   @Test
 133   public void testScanBatch() throws Exception {
 134     final TableName tableName = TableName.valueOf(name.getMethodName());
 135     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 8);
 136
 137     Table ht = TEST_UTIL.createTable(tableName, FAMILY);
 138
 139     Put put;
 140     Scan scan;
 141     Delete delete;
 142     Result result;
 143     ResultScanner scanner;
 144     boolean toLog = true;
 145     List<Cell> kvListExp;
 146
 147     // table: row, family, c0:0, c1:1, ... , c7:7
 148     put = new Put(ROW);
 149     for (int i=0; i < QUALIFIERS.length; i++) {
 150       KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
 151       put.add(kv);
 152     }
 153     ht.put(put);
 154
 155     // table: row, family, c0:0, c1:1, ..., c6:2, c6:6 , c7:7
 156     put = new Put(ROW);
 157     KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[6], 2, VALUE);
 158     put.add(kv);
 159     ht.put(put);
 160
 161     // delete upto ts: 3
 162     delete = new Delete(ROW);
 163     delete.addFamily(FAMILY, 3);
 164     ht.delete(delete);
 165
 166     // without batch
 167     scan = new Scan().withStartRow(ROW);
 168     scan.setMaxVersions();
 169     scanner = ht.getScanner(scan);
 170
 171     // c4:4, c5:5, c6:6, c7:7
 172     kvListExp = new ArrayList<>();
 173     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
 174     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
 175     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
 176     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
 177     result = scanner.next();
 178     verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
 179
 180     // with batch
 181     scan =  new Scan().withStartRow(ROW);
 182     scan.setMaxVersions();
 183     scan.setBatch(2);
 184     scanner = ht.getScanner(scan);
 185
 186     // First batch: c4:4, c5:5
 187     kvListExp = new ArrayList<>();
 188     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
 189     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
 190     result = scanner.next();
 191     verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
 192
 193     // Second batch: c6:6, c7:7
 194     kvListExp = new ArrayList<>();
 195     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
 196     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
 197     result = scanner.next();
 198     verifyResult(result, kvListExp, toLog, "Testing second batch of scan");
 199
 200   }
 201
 202   @Test
 203   public void testMaxResultSizeIsSetToDefault() throws Exception {
 204     final TableName tableName = TableName.valueOf(name.getMethodName());
 205     Table ht = TEST_UTIL.createTable(tableName, FAMILY);
 206
 207     // The max result size we expect the scan to use by default.
 208     long expectedMaxResultSize =
 209         TEST_UTIL.getConfiguration().getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
 210           HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
 211
 212     int numRows = 5;
 213     byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
 214
 215     int numQualifiers = 10;
 216     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
 217
 218     // Specify the cell size such that a single row will be larger than the default
 219     // value of maxResultSize. This means that Scan RPCs should return at most a single
 220     // result back to the client.
 221     int cellSize = (int) (expectedMaxResultSize / (numQualifiers - 1));
 222     byte[] cellValue = Bytes.createMaxByteArray(cellSize);
 223
 224     Put put;
 225     List<Put> puts = new ArrayList<>();
 226     for (int row = 0; row < ROWS.length; row++) {
 227       put = new Put(ROWS[row]);
 228       for (int qual = 0; qual < QUALIFIERS.length; qual++) {
 229         KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], cellValue);
 230         put.add(kv);
 231       }
 232       puts.add(put);
 233     }
 234     ht.put(puts);
 235
 236     // Create a scan with the default configuration.
 237     Scan scan = new Scan();
 238
 239     ResultScanner scanner = ht.getScanner(scan);
 240     assertTrue(scanner instanceof ClientScanner);
 241     ClientScanner clientScanner = (ClientScanner) scanner;
 242
 243     // Call next to issue a single RPC to the server
 244     scanner.next();
 245
 246     // The scanner should have, at most, a single result in its cache. If there more results exists
 247     // in the cache it means that more than the expected max result size was fetched.
 248     assertTrue("The cache contains: " + clientScanner.getCacheSize() + " results",
 249       clientScanner.getCacheSize() <= 1);
 250   }
 251
 252   /**
 253    * Scan on not existing table should throw the exception with correct message
 254    */
 255   @Test
 256   public void testScannerForNotExistingTable() {
 257     String[] tableNames = {"A", "Z", "A:A", "Z:Z"};
 258     for(String tableName : tableNames) {
 259       try {
 260         Table table = TEST_UTIL.getConnection().getTable(TableName.valueOf(tableName));
 261         testSmallScan(table, true, 1, 5);
 262         fail("TableNotFoundException was not thrown");
 263       } catch (TableNotFoundException e) {
 264         // We expect that the message for TableNotFoundException would have only the table name only
 265         // Otherwise that would mean that localeRegionInMeta doesn't work properly
 266         assertEquals(e.getMessage(), tableName);
 267       } catch (Exception e) {
 268         fail("Unexpected exception " + e.getMessage());
 269       }
 270     }
 271   }
 272
 273   @Test
 274   public void testSmallScan() throws Exception {
 275     final TableName tableName = TableName.valueOf(name.getMethodName());
 276
 277     int numRows = 10;
 278     byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
 279
 280     int numQualifiers = 10;
 281     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
 282
 283     Table ht = TEST_UTIL.createTable(tableName, FAMILY);
 284
 285     Put put;
 286     List<Put> puts = new ArrayList<>();
 287     for (int row = 0; row < ROWS.length; row++) {
 288       put = new Put(ROWS[row]);
 289       for (int qual = 0; qual < QUALIFIERS.length; qual++) {
 290         KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], VALUE);
 291         put.add(kv);
 292       }
 293       puts.add(put);
 294     }
 295     ht.put(puts);
 296
 297     int expectedRows = numRows;
 298     int expectedCols = numRows * numQualifiers;
 299
 300     // Test normal and reversed
 301     testSmallScan(ht, true, expectedRows, expectedCols);
 302     testSmallScan(ht, false, expectedRows, expectedCols);
 303   }
 304
 305   /**
 306    * Run through a variety of test configurations with a small scan
 307    * @param table
 308    * @param reversed
 309    * @param rows
 310    * @param columns
 311    * @throws Exception
 312    */
 313   private void testSmallScan(Table table, boolean reversed, int rows, int columns) throws Exception {
 314     Scan baseScan = new Scan();
 315     baseScan.setReversed(reversed);
 316     baseScan.setSmall(true);
 317
 318     Scan scan = new Scan(baseScan);
 319     verifyExpectedCounts(table, scan, rows, columns);
 320
 321     scan = new Scan(baseScan);
 322     scan.setMaxResultSize(1);
 323     verifyExpectedCounts(table, scan, rows, columns);
 324
 325     scan = new Scan(baseScan);
 326     scan.setMaxResultSize(1);
 327     scan.setCaching(Integer.MAX_VALUE);
 328     verifyExpectedCounts(table, scan, rows, columns);
 329   }
 330
 331   private void verifyExpectedCounts(Table table, Scan scan, int expectedRowCount,
 332       int expectedCellCount) throws Exception {
 333     ResultScanner scanner = table.getScanner(scan);
 334
 335     int rowCount = 0;
 336     int cellCount = 0;
 337     Result r = null;
 338     while ((r = scanner.next()) != null) {
 339       rowCount++;
 340       cellCount += r.rawCells().length;
 341     }
 342
 343     assertTrue("Expected row count: " + expectedRowCount + " Actual row count: " + rowCount,
 344         expectedRowCount == rowCount);
 345     assertTrue("Expected cell count: " + expectedCellCount + " Actual cell count: " + cellCount,
 346         expectedCellCount == cellCount);
 347     scanner.close();
 348   }
 349
 350   /**
 351    * Test from client side for get with maxResultPerCF set
 352    *
 353    * @throws Exception
 354    */
 355   @Test
 356   public void testGetMaxResults() throws Exception {
 357     final TableName tableName = TableName.valueOf(name.getMethodName());
 358     byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
 359     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
 360
 361     Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
 362
 363     Get get;
 364     Put put;
 365     Result result;
 366     boolean toLog = true;
 367     List<Cell> kvListExp;
 368
 369     kvListExp = new ArrayList<>();
 370     // Insert one CF for row[0]
 371     put = new Put(ROW);
 372     for (int i=0; i < 10; i++) {
 373       KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
 374       put.add(kv);
 375       kvListExp.add(kv);
 376     }
 377     ht.put(put);
 378
 379     get = new Get(ROW);
 380     result = ht.get(get);
 381     verifyResult(result, kvListExp, toLog, "Testing without setting maxResults");
 382
 383     get = new Get(ROW);
 384     get.setMaxResultsPerColumnFamily(2);
 385     result = ht.get(get);
 386     kvListExp = new ArrayList<>();
 387     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[0], 1, VALUE));
 388     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
 389     verifyResult(result, kvListExp, toLog, "Testing basic setMaxResults");
 390
 391     // Filters: ColumnRangeFilter
 392     get = new Get(ROW);
 393     get.setMaxResultsPerColumnFamily(5);
 394     get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5],
 395                                         true));
 396     result = ht.get(get);
 397     kvListExp = new ArrayList<>();
 398     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[2], 1, VALUE));
 399     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
 400     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
 401     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
 402     verifyResult(result, kvListExp, toLog, "Testing single CF with CRF");
 403
 404     // Insert two more CF for row[0]
 405     // 20 columns for CF2, 10 columns for CF1
 406     put = new Put(ROW);
 407     for (int i=0; i < QUALIFIERS.length; i++) {
 408       KeyValue kv = new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE);
 409       put.add(kv);
 410     }
 411     ht.put(put);
 412
 413     put = new Put(ROW);
 414     for (int i=0; i < 10; i++) {
 415       KeyValue kv = new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE);
 416       put.add(kv);
 417     }
 418     ht.put(put);
 419
 420     get = new Get(ROW);
 421     get.setMaxResultsPerColumnFamily(12);
 422     get.addFamily(FAMILIES[1]);
 423     get.addFamily(FAMILIES[2]);
 424     result = ht.get(get);
 425     kvListExp = new ArrayList<>();
 426     //Exp: CF1:q0, ..., q9, CF2: q0, q1, q10, q11, ..., q19
 427     for (int i=0; i < 10; i++) {
 428       kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
 429     }
 430     for (int i=0; i < 2; i++) {
 431         kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
 432       }
 433     for (int i=10; i < 20; i++) {
 434       kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
 435     }
 436     verifyResult(result, kvListExp, toLog, "Testing multiple CFs");
 437
 438     // Filters: ColumnRangeFilter and ColumnPrefixFilter
 439     get = new Get(ROW);
 440     get.setMaxResultsPerColumnFamily(3);
 441     get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, null, true));
 442     result = ht.get(get);
 443     kvListExp = new ArrayList<>();
 444     for (int i=2; i < 5; i++) {
 445       kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
 446     }
 447     for (int i=2; i < 5; i++) {
 448       kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
 449     }
 450     for (int i=2; i < 5; i++) {
 451       kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
 452     }
 453     verifyResult(result, kvListExp, toLog, "Testing multiple CFs + CRF");
 454
 455     get = new Get(ROW);
 456     get.setMaxResultsPerColumnFamily(7);
 457     get.setFilter(new ColumnPrefixFilter(QUALIFIERS[1]));
 458     result = ht.get(get);
 459     kvListExp = new ArrayList<>();
 460     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
 461     kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[1], 1, VALUE));
 462     kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[1], 1, VALUE));
 463     for (int i=10; i < 16; i++) {
 464       kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
 465     }
 466     verifyResult(result, kvListExp, toLog, "Testing multiple CFs + PFF");
 467
 468   }
 469
 470   /**
 471    * Test from client side for scan with maxResultPerCF set
 472    *
 473    * @throws Exception
 474    */
 475   @Test
 476   public void testScanMaxResults() throws Exception {
 477     final TableName tableName = TableName.valueOf(name.getMethodName());
 478     byte [][] ROWS = HTestConst.makeNAscii(ROW, 2);
 479     byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
 480     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 10);
 481
 482     Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
 483
 484     Put put;
 485     Scan scan;
 486     Result result;
 487     boolean toLog = true;
 488     List<Cell> kvListExp, kvListScan;
 489
 490     kvListExp = new ArrayList<>();
 491
 492     for (int r=0; r < ROWS.length; r++) {
 493       put = new Put(ROWS[r]);
 494       for (int c=0; c < FAMILIES.length; c++) {
 495         for (int q=0; q < QUALIFIERS.length; q++) {
 496           KeyValue kv = new KeyValue(ROWS[r], FAMILIES[c], QUALIFIERS[q], 1, VALUE);
 497           put.add(kv);
 498           if (q < 4) {
 499             kvListExp.add(kv);
 500           }
 501         }
 502       }
 503       ht.put(put);
 504     }
 505
 506     scan = new Scan();
 507     scan.setMaxResultsPerColumnFamily(4);
 508     ResultScanner scanner = ht.getScanner(scan);
 509     kvListScan = new ArrayList<>();
 510     while ((result = scanner.next()) != null) {
 511       for (Cell kv : result.listCells()) {
 512         kvListScan.add(kv);
 513       }
 514     }
 515     result = Result.create(kvListScan);
 516     verifyResult(result, kvListExp, toLog, "Testing scan with maxResults");
 517
 518   }
 519
 520   /**
 521    * Test from client side for get with rowOffset
 522    *
 523    * @throws Exception
 524    */
 525   @Test
 526   public void testGetRowOffset() throws Exception {
 527     final TableName tableName = TableName.valueOf(name.getMethodName());
 528     byte [][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
 529     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
 530
 531     Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
 532
 533     Get get;
 534     Put put;
 535     Result result;
 536     boolean toLog = true;
 537     List<Cell> kvListExp;
 538
 539     // Insert one CF for row
 540     kvListExp = new ArrayList<>();
 541     put = new Put(ROW);
 542     for (int i=0; i < 10; i++) {
 543       KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
 544       put.add(kv);
 545       // skipping first two kvs
 546       if (i < 2) continue;
 547       kvListExp.add(kv);
 548     }
 549     ht.put(put);
 550
 551     //setting offset to 2
 552     get = new Get(ROW);
 553     get.setRowOffsetPerColumnFamily(2);
 554     result = ht.get(get);
 555     verifyResult(result, kvListExp, toLog, "Testing basic setRowOffset");
 556
 557     //setting offset to 20
 558     get = new Get(ROW);
 559     get.setRowOffsetPerColumnFamily(20);
 560     result = ht.get(get);
 561     kvListExp = new ArrayList<>();
 562     verifyResult(result, kvListExp, toLog, "Testing offset > #kvs");
 563
 564     //offset + maxResultPerCF
 565     get = new Get(ROW);
 566     get.setRowOffsetPerColumnFamily(4);
 567     get.setMaxResultsPerColumnFamily(5);
 568     result = ht.get(get);
 569     kvListExp = new ArrayList<>();
 570     for (int i=4; i < 9; i++) {
 571       kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
 572     }
 573     verifyResult(result, kvListExp, toLog,
 574       "Testing offset + setMaxResultsPerCF");
 575
 576     // Filters: ColumnRangeFilter
 577     get = new Get(ROW);
 578     get.setRowOffsetPerColumnFamily(1);
 579     get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5],
 580                                         true));
 581     result = ht.get(get);
 582     kvListExp = new ArrayList<>();
 583     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
 584     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
 585     kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
 586     verifyResult(result, kvListExp, toLog, "Testing offset with CRF");
 587
 588     // Insert into two more CFs for row
 589     // 10 columns for CF2, 10 columns for CF1
 590     for(int j=2; j > 0; j--) {
 591       put = new Put(ROW);
 592       for (int i=0; i < 10; i++) {
 593         KeyValue kv = new KeyValue(ROW, FAMILIES[j], QUALIFIERS[i], 1, VALUE);
 594         put.add(kv);
 595       }
 596       ht.put(put);
 597     }
 598
 599     get = new Get(ROW);
 600     get.setRowOffsetPerColumnFamily(4);
 601     get.setMaxResultsPerColumnFamily(2);
 602     get.addFamily(FAMILIES[1]);
 603     get.addFamily(FAMILIES[2]);
 604     result = ht.get(get);
 605     kvListExp = new ArrayList<>();
 606     //Exp: CF1:q4, q5, CF2: q4, q5
 607     kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[4], 1, VALUE));
 608     kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[5], 1, VALUE));
 609     kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[4], 1, VALUE));
 610     kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[5], 1, VALUE));
 611     verifyResult(result, kvListExp, toLog,
 612        "Testing offset + multiple CFs + maxResults");
 613   }
 614
 615   @Test
 616   public void testScanRawDeleteFamilyVersion() throws Exception {
 617     TableName tableName = TableName.valueOf(name.getMethodName());
 618     TEST_UTIL.createTable(tableName, FAMILY);
 619     Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
 620     conf.set(RPC_CODEC_CONF_KEY, "");
 621     conf.set(DEFAULT_CODEC_CLASS, "");
 622     try (Connection connection = ConnectionFactory.createConnection(conf);
 623         Table table = connection.getTable(tableName)) {
 624       Delete delete = new Delete(ROW);
 625       delete.addFamilyVersion(FAMILY, 0L);
 626       table.delete(delete);
 627       Scan scan = new Scan(ROW).setRaw(true);
 628       ResultScanner scanner = table.getScanner(scan);
 629       int count = 0;
 630       while (scanner.next() != null) {
 631         count++;
 632       }
 633       assertEquals(1, count);
 634     } finally {
 635       TEST_UTIL.deleteTable(tableName);
 636     }
 637   }
 638
 639   /**
 640    * Test from client side for scan while the region is reopened
 641    * on the same region server.
 642    *
 643    * @throws Exception
 644    */
 645   @Test
 646   public void testScanOnReopenedRegion() throws Exception {
 647     final TableName tableName = TableName.valueOf(name.getMethodName());
 648     byte [][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 2);
 649
 650     Table ht = TEST_UTIL.createTable(tableName, FAMILY);
 651
 652     Put put;
 653     Scan scan;
 654     Result result;
 655     ResultScanner scanner;
 656     boolean toLog = false;
 657     List<Cell> kvListExp;
 658
 659     // table: row, family, c0:0, c1:1
 660     put = new Put(ROW);
 661     for (int i=0; i < QUALIFIERS.length; i++) {
 662       KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
 663       put.add(kv);
 664     }
 665     ht.put(put);
 666
 667     scan = new Scan().withStartRow(ROW);
 668     scanner = ht.getScanner(scan);
 669
 670     HRegionLocation loc;
 671
 672     try (RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName)) {
 673       loc = locator.getRegionLocation(ROW);
 674     }
 675     HRegionInfo hri = loc.getRegionInfo();
 676     MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
 677     byte[] regionName = hri.getRegionName();
 678     int i = cluster.getServerWith(regionName);
 679     HRegionServer rs = cluster.getRegionServer(i);
 680     LOG.info("Unassigning " + hri);
 681     TEST_UTIL.getAdmin().unassign(hri.getRegionName(), true);
 682     long startTime = EnvironmentEdgeManager.currentTime();
 683     long timeOut = 10000;
 684     boolean offline = false;
 685     while (true) {
 686       if (rs.getOnlineRegion(regionName) == null) {
 687         offline = true;
 688         break;
 689       }
 690       assertTrue("Timed out in closing the testing region",
 691         EnvironmentEdgeManager.currentTime() < startTime + timeOut);
 692     }
 693     assertTrue(offline);
 694     LOG.info("Assigning " + hri);
 695     TEST_UTIL.getAdmin().assign(hri.getRegionName());
 696     startTime = EnvironmentEdgeManager.currentTime();
 697     while (true) {
 698       rs = cluster.getRegionServer(cluster.getServerWith(regionName));
 699       if (rs != null && rs.getOnlineRegion(regionName) != null) {
 700         offline = false;
 701         break;
 702       }
 703       assertTrue("Timed out in open the testing region",
 704         EnvironmentEdgeManager.currentTime() < startTime + timeOut);
 705     }
 706     assertFalse(offline);
 707
 708     // c0:0, c1:1
 709     kvListExp = new ArrayList<>();
 710     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[0], 0, VALUE));
 711     kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[1], 1, VALUE));
 712     result = scanner.next();
 713     verifyResult(result, kvListExp, toLog, "Testing scan on re-opened region");
 714   }
 715
 716   @Test
 717   public void testAsyncScannerWithSmallData() throws Exception {
 718     testAsyncScanner(TableName.valueOf(name.getMethodName()),
 719       2,
 720       3,
 721       10,
 722       -1,
 723       null);
 724   }
 725
 726   @Test
 727   public void testAsyncScannerWithManyRows() throws Exception {
 728     testAsyncScanner(TableName.valueOf(name.getMethodName()),
 729       30000,
 730       1,
 731       1,
 732       -1,
 733       null);
 734   }
 735
 736   @Test
 737   public void testAsyncScannerWithoutCaching() throws Exception {
 738     testAsyncScanner(TableName.valueOf(name.getMethodName()),
 739       5,
 740       1,
 741       1,
 742       1,
 743       (b) -> {
 744         try {
 745           TimeUnit.MILLISECONDS.sleep(500);
 746         } catch (InterruptedException ex) {
 747         }
 748       });
 749   }
 750
 751   private void testAsyncScanner(TableName table, int rowNumber, int familyNumber,
 752       int qualifierNumber, int caching, Consumer<Boolean> listener) throws Exception {
 753     assert rowNumber > 0;
 754     assert familyNumber > 0;
 755     assert qualifierNumber > 0;
 756     byte[] row = Bytes.toBytes("r");
 757     byte[] family = Bytes.toBytes("f");
 758     byte[] qualifier = Bytes.toBytes("q");
 759     byte[][] rows = makeNAsciiWithZeroPrefix(row, rowNumber);
 760     byte[][] families = makeNAsciiWithZeroPrefix(family, familyNumber);
 761     byte[][] qualifiers = makeNAsciiWithZeroPrefix(qualifier, qualifierNumber);
 762
 763     Table ht = TEST_UTIL.createTable(table, families);
 764
 765     boolean toLog = true;
 766     List<Cell> kvListExp = new ArrayList<>();
 767
 768     List<Put> puts = new ArrayList<>();
 769     for (byte[] r : rows) {
 770       Put put = new Put(r);
 771       for (byte[] f : families) {
 772         for (byte[] q : qualifiers) {
 773           KeyValue kv = new KeyValue(r, f, q, 1, VALUE);
 774           put.add(kv);
 775           kvListExp.add(kv);
 776         }
 777       }
 778       puts.add(put);
 779       if (puts.size() > 1000) {
 780         ht.put(puts);
 781         puts.clear();
 782       }
 783     }
 784     if (!puts.isEmpty()) {
 785       ht.put(puts);
 786       puts.clear();
 787     }
 788
 789     Scan scan = new Scan();
 790     scan.setAsyncPrefetch(true);
 791     if (caching > 0) {
 792       scan.setCaching(caching);
 793     }
 794     try (ResultScanner scanner = ht.getScanner(scan)) {
 795       assertTrue("Not instance of async scanner",scanner instanceof ClientAsyncPrefetchScanner);
 796       ((ClientAsyncPrefetchScanner) scanner).setPrefetchListener(listener);
 797       List<Cell> kvListScan = new ArrayList<>();
 798       Result result;
 799       boolean first = true;
 800       int actualRows = 0;
 801       while ((result = scanner.next()) != null) {
 802         ++actualRows;
 803         // waiting for cache. see HBASE-17376
 804         if (first) {
 805           TimeUnit.SECONDS.sleep(1);
 806           first = false;
 807         }
 808         for (Cell kv : result.listCells()) {
 809           kvListScan.add(kv);
 810         }
 811       }
 812       assertEquals(rowNumber, actualRows);
 813       // These cells may have different rows but it is ok. The Result#getRow
 814       // isn't used in the verifyResult()
 815       result = Result.create(kvListScan);
 816       verifyResult(result, kvListExp, toLog, "Testing async scan");
 817     }
 818
 819     TEST_UTIL.deleteTable(table);
 820   }
 821
 822   private static byte[][] makeNAsciiWithZeroPrefix(byte[] base, int n) {
 823     int maxLength = Integer.toString(n).length();
 824     byte [][] ret = new byte[n][];
 825     for (int i = 0; i < n; i++) {
 826       int length = Integer.toString(i).length();
 827       StringBuilder buf = new StringBuilder(Integer.toString(i));
 828       IntStream.range(0, maxLength - length).forEach(v -> buf.insert(0, "0"));
 829       byte[] tail = Bytes.toBytes(buf.toString());
 830       ret[i] = Bytes.add(base, tail);
 831     }
 832     return ret;
 833   }
 834
 835   static void verifyResult(Result result, List<Cell> expKvList, boolean toLog,
 836       String msg) {
 837
 838     LOG.info(msg);
 839     LOG.info("Expected count: " + expKvList.size());
 840     LOG.info("Actual count: " + result.size());
 841     if (expKvList.isEmpty())
 842       return;
 843
 844     int i = 0;
 845     for (Cell kv : result.rawCells()) {
 846       if (i >= expKvList.size()) {
 847         break;  // we will check the size later
 848       }
 849
 850       Cell kvExp = expKvList.get(i++);
 851       if (toLog) {
 852         LOG.info("get kv is: " + kv.toString());
 853         LOG.info("exp kv is: " + kvExp.toString());
 854       }
 855       assertTrue("Not equal", kvExp.equals(kv));
 856     }
 857
 858     assertEquals(expKvList.size(), result.size());
 859   }
 860
 861   @Test
 862   public void testReadExpiredDataForRawScan() throws IOException {
 863     TableName tableName = TableName.valueOf(name.getMethodName());
 864     long ts = System.currentTimeMillis() - 10000;
 865     byte[] value = Bytes.toBytes("expired");
 866     try (Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
 867       table.put(new Put(ROW).addColumn(FAMILY, QUALIFIER, ts, value));
 868       assertArrayEquals(value, table.get(new Get(ROW)).getValue(FAMILY, QUALIFIER));
 869       TEST_UTIL.getAdmin().modifyColumnFamily(tableName,
 870         new HColumnDescriptor(FAMILY).setTimeToLive(5));
 871       try (ResultScanner scanner = table.getScanner(FAMILY)) {
 872         assertNull(scanner.next());
 873       }
 874       try (ResultScanner scanner = table.getScanner(new Scan().setRaw(true))) {
 875         assertArrayEquals(value, scanner.next().getValue(FAMILY, QUALIFIER));
 876         assertNull(scanner.next());
 877       }
 878     }
 879   }
 880
 881   @Test
 882   public void testScanWithColumnsAndFilterAndVersion() throws IOException {
 883     TableName tableName = TableName.valueOf(name.getMethodName());
 884     try (Table table = TEST_UTIL.createTable(tableName, FAMILY, 4)) {
 885       for (int i = 0; i < 4; i++) {
 886         Put put = new Put(ROW);
 887         put.addColumn(FAMILY, QUALIFIER, VALUE);
 888         table.put(put);
 889       }
 890
 891       Scan scan = new Scan();
 892       scan.addColumn(FAMILY, QUALIFIER);
 893       scan.setFilter(new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(QUALIFIER)));
 894       scan.readVersions(3);
 895
 896       try (ResultScanner scanner = table.getScanner(scan)) {
 897         Result result = scanner.next();
 898         assertEquals(3, result.size());
 899       }
 900     }
 901   }
 902
 903   @Test
 904   public void testScanWithSameStartRowStopRow() throws IOException {
 905     TableName tableName = TableName.valueOf(name.getMethodName());
 906     try (Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
 907       table.put(new Put(ROW).addColumn(FAMILY, QUALIFIER, VALUE));
 908
 909       Scan scan = new Scan().withStartRow(ROW).withStopRow(ROW);
 910       try (ResultScanner scanner = table.getScanner(scan)) {
 911         assertNull(scanner.next());
 912       }
 913
 914       scan = new Scan().withStartRow(ROW, true).withStopRow(ROW, true);
 915       try (ResultScanner scanner = table.getScanner(scan)) {
 916         Result result = scanner.next();
 917         assertNotNull(result);
 918         assertArrayEquals(ROW, result.getRow());
 919         assertArrayEquals(VALUE, result.getValue(FAMILY, QUALIFIER));
 920         assertNull(scanner.next());
 921       }
 922
 923       scan = new Scan().withStartRow(ROW, true).withStopRow(ROW, false);
 924       try (ResultScanner scanner = table.getScanner(scan)) {
 925         assertNull(scanner.next());
 926       }
 927
 928       scan = new Scan().withStartRow(ROW, false).withStopRow(ROW, false);
 929       try (ResultScanner scanner = table.getScanner(scan)) {
 930         assertNull(scanner.next());
 931       }
 932
 933       scan = new Scan().withStartRow(ROW, false).withStopRow(ROW, true);
 934       try (ResultScanner scanner = table.getScanner(scan)) {
 935         assertNull(scanner.next());
 936       }
 937     }
 938   }
 939
 940   @Test
 941   public void testReverseScanWithFlush() throws Exception {
 942     TableName tableName = TableName.valueOf(name.getMethodName());
 943     final int BATCH_SIZE = 10;
 944     final int ROWS_TO_INSERT = 100;
 945     final byte[] LARGE_VALUE = generateHugeValue(128 * 1024);
 946
 947     try (Table table = TEST_UTIL.createTable(tableName, FAMILY);
 948         Admin admin = TEST_UTIL.getAdmin()) {
 949       List<Put> putList = new ArrayList<>();
 950       for (long i = 0; i < ROWS_TO_INSERT; i++) {
 951         Put put = new Put(Bytes.toBytes(i));
 952         put.addColumn(FAMILY, QUALIFIER, LARGE_VALUE);
 953         putList.add(put);
 954
 955         if (putList.size() >= BATCH_SIZE) {
 956           table.put(putList);
 957           admin.flush(tableName);
 958           putList.clear();
 959         }
 960       }
 961
 962       if (!putList.isEmpty()) {
 963         table.put(putList);
 964         admin.flush(tableName);
 965         putList.clear();
 966       }
 967
 968       Scan scan = new Scan();
 969       scan.setReversed(true);
 970       int count = 0;
 971
 972       try (ResultScanner results = table.getScanner(scan)) {
 973         for (Result result : results) {
 974           count++;
 975         }
 976       }
 977       assertEquals("Expected " + ROWS_TO_INSERT + " rows in the table but it is " + count,
 978           ROWS_TO_INSERT, count);
 979     }
 980   }
 981 }