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