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