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