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.regionserver; 019 020import static org.apache.hadoop.hbase.HBaseTestingUtil.START_KEY_BYTES; 021import static org.apache.hadoop.hbase.HBaseTestingUtil.fam1; 022import static org.apache.hadoop.hbase.HBaseTestingUtil.fam2; 023import static org.junit.jupiter.api.Assertions.assertEquals; 024import static org.junit.jupiter.api.Assertions.assertFalse; 025import static org.junit.jupiter.api.Assertions.assertNotNull; 026import static org.junit.jupiter.api.Assertions.assertTrue; 027import static org.junit.jupiter.api.Assertions.fail; 028 029import java.io.IOException; 030import java.util.ArrayList; 031import java.util.List; 032import java.util.NavigableSet; 033import org.apache.hadoop.hbase.Cell; 034import org.apache.hadoop.hbase.CellUtil; 035import org.apache.hadoop.hbase.CompareOperator; 036import org.apache.hadoop.hbase.ExtendedCell; 037import org.apache.hadoop.hbase.HBaseTestingUtil; 038import org.apache.hadoop.hbase.HConstants; 039import org.apache.hadoop.hbase.HTestConst; 040import org.apache.hadoop.hbase.TableName; 041import org.apache.hadoop.hbase.UnknownScannerException; 042import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 043import org.apache.hadoop.hbase.client.Delete; 044import org.apache.hadoop.hbase.client.Get; 045import org.apache.hadoop.hbase.client.Put; 046import org.apache.hadoop.hbase.client.RegionInfo; 047import org.apache.hadoop.hbase.client.RegionInfoBuilder; 048import org.apache.hadoop.hbase.client.Result; 049import org.apache.hadoop.hbase.client.ResultScanner; 050import org.apache.hadoop.hbase.client.Scan; 051import org.apache.hadoop.hbase.client.Table; 052import org.apache.hadoop.hbase.client.TableDescriptor; 053import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 054import org.apache.hadoop.hbase.filter.BinaryComparator; 055import org.apache.hadoop.hbase.filter.ByteArrayComparable; 056import org.apache.hadoop.hbase.filter.Filter; 057import org.apache.hadoop.hbase.filter.InclusiveStopFilter; 058import org.apache.hadoop.hbase.filter.PrefixFilter; 059import org.apache.hadoop.hbase.filter.RowFilter; 060import org.apache.hadoop.hbase.filter.WhileMatchFilter; 061import org.apache.hadoop.hbase.testclassification.MediumTests; 062import org.apache.hadoop.hbase.testclassification.RegionServerTests; 063import org.apache.hadoop.hbase.util.Bytes; 064import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 065import org.junit.jupiter.api.Tag; 066import org.junit.jupiter.api.Test; 067import org.junit.jupiter.api.TestInfo; 068import org.slf4j.Logger; 069import org.slf4j.LoggerFactory; 070 071/** 072 * Test of a long-lived scanner validating as we go. 073 */ 074@Tag(RegionServerTests.TAG) 075@Tag(MediumTests.TAG) 076public class TestScanner { 077 078 private static final Logger LOG = LoggerFactory.getLogger(TestScanner.class); 079 private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 080 081 private static final byte[] FIRST_ROW = HConstants.EMPTY_START_ROW; 082 private static final byte[][] COLS = { HConstants.CATALOG_FAMILY }; 083 private static final byte[][] EXPLICIT_COLS = 084 { HConstants.REGIONINFO_QUALIFIER, HConstants.SERVER_QUALIFIER, 085 // TODO ryan 086 // HConstants.STARTCODE_QUALIFIER 087 }; 088 089 static final TableDescriptor TESTTABLEDESC = 090 TableDescriptorBuilder.newBuilder(TableName.valueOf("testscanner")) 091 .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(HConstants.CATALOG_FAMILY) 092 // Ten is an arbitrary number. Keep versions to help debugging. 093 .setMaxVersions(10).setBlockCacheEnabled(false).setBlocksize(8 * 1024).build()) 094 .build(); 095 096 /** HRegionInfo for root region */ 097 public static final RegionInfo REGION_INFO = 098 RegionInfoBuilder.newBuilder(TESTTABLEDESC.getTableName()).build(); 099 100 private static final byte[] ROW_KEY = REGION_INFO.getRegionName(); 101 102 private static final long START_CODE = Long.MAX_VALUE; 103 104 private static final byte[] LAZY_SEEK_FAMILY = Bytes.toBytes("family"); 105 private static final byte[] LAZY_SEEK_QUALIFIER = Bytes.toBytes("qualifier"); 106 private static final byte[] LAZY_SEEK_ROW = Bytes.toBytes("row"); 107 108 private HRegion region; 109 110 private byte[] firstRowBytes, secondRowBytes, thirdRowBytes; 111 final private byte[] col1; 112 113 public TestScanner() { 114 super(); 115 116 firstRowBytes = START_KEY_BYTES; 117 secondRowBytes = START_KEY_BYTES.clone(); 118 // Increment the least significant character so we get to next row. 119 secondRowBytes[START_KEY_BYTES.length - 1]++; 120 thirdRowBytes = START_KEY_BYTES.clone(); 121 thirdRowBytes[START_KEY_BYTES.length - 1] = 122 (byte) (thirdRowBytes[START_KEY_BYTES.length - 1] + 2); 123 col1 = Bytes.toBytes("column1"); 124 } 125 126 private static final class RecordingStoreScanner extends StoreScanner { 127 private boolean initialSeekWasLazy; 128 129 RecordingStoreScanner(HStore store, Scan scan, NavigableSet<byte[]> columns) 130 throws IOException { 131 super(store, store.getScanInfo(), scan, columns, Long.MAX_VALUE); 132 } 133 134 @Override 135 protected void seekScanners(List<? extends KeyValueScanner> scanners, ExtendedCell seekKey, 136 boolean isLazy, boolean isParallelSeek) throws IOException { 137 initialSeekWasLazy = isLazy; 138 super.seekScanners(scanners, seekKey, isLazy, isParallelSeek); 139 } 140 } 141 142 private static final class TrackingRowComparator extends ByteArrayComparable { 143 private final List<byte[]> comparedRows = new ArrayList<>(); 144 145 TrackingRowComparator(byte[] value) { 146 super(value); 147 } 148 149 @Override 150 public int compareTo(byte[] value, int offset, int length) { 151 comparedRows.add(Bytes.copy(value, offset, length)); 152 return Bytes.compareTo(getValue(), 0, getValue().length, value, offset, length); 153 } 154 155 @Override 156 public byte[] toByteArray() { 157 return getValue(); 158 } 159 } 160 161 @Test 162 public void testFilterComparatorOnlySeesActualRows() throws Exception { 163 byte[] family = Bytes.toBytes("family"); 164 byte[] qualifier = Bytes.toBytes("qualifier"); 165 byte[] regionStartKey = new byte[] { 1 }; 166 byte[] row = new byte[] { 1, 0, 1 }; 167 TableDescriptor tableDescriptor = 168 TableDescriptorBuilder.newBuilder(TableName.valueOf("testFilterComparatorOnlySeesActualRows")) 169 .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family) 170 .setBloomFilterType(BloomType.ROWCOL).build()) 171 .build(); 172 TrackingRowComparator comparator = new TrackingRowComparator(row); 173 174 StoreScanner.enableLazySeekGlobally(true); 175 try { 176 this.region = TEST_UTIL.createLocalHRegion(tableDescriptor, regionStartKey, null); 177 Put put = new Put(row); 178 put.addColumn(family, qualifier, Bytes.toBytes("value")); 179 region.put(put); 180 region.flush(true); 181 182 Scan scan = new Scan().withStartRow(regionStartKey); 183 scan.addColumn(family, qualifier); 184 scan.setFilter(new RowFilter(CompareOperator.EQUAL, comparator)); 185 List<Cell> results = new ArrayList<>(); 186 try (InternalScanner scanner = region.getScanner(scan)) { 187 assertFalse(scanner.next(results)); 188 } 189 190 assertEquals(1, results.size()); 191 assertTrue(CellUtil.matchingRows(results.get(0), row)); 192 assertEquals(1, comparator.comparedRows.size()); 193 assertTrue(Bytes.equals(row, comparator.comparedRows.get(0))); 194 } finally { 195 StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT); 196 HBaseTestingUtil.closeRegionAndWAL(this.region); 197 } 198 } 199 200 @Test 201 public void testWhileMatchFilterOnlySeesActualRows() throws Exception { 202 byte[] family = Bytes.toBytes("family"); 203 byte[] qualifier = Bytes.toBytes("qualifier"); 204 TableDescriptor tableDescriptor = 205 TableDescriptorBuilder.newBuilder(TableName.valueOf("testWhileMatchFilterOnlySeesActualRows")) 206 .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family).build()).build(); 207 208 StoreScanner.enableLazySeekGlobally(true); 209 try { 210 this.region = TEST_UTIL.createLocalHRegion(tableDescriptor, null, null); 211 List<String> rows = List.of("row1", "row2", "row3"); 212 for (String row : rows) { 213 Put put = new Put(Bytes.toBytes(row)).addColumn(family, qualifier, Bytes.toBytes("value")); 214 region.put(put); 215 } 216 region.flush(true); 217 218 Scan scan = new Scan().addColumn(family, qualifier) 219 .setFilter(new WhileMatchFilter(new RowFilter(CompareOperator.NOT_EQUAL, 220 new BinaryComparator(HConstants.EMPTY_START_ROW)))); 221 int scannedRows = 0; 222 try (InternalScanner scanner = region.getScanner(scan)) { 223 boolean hasMoreRows; 224 do { 225 List<Cell> results = new ArrayList<>(); 226 hasMoreRows = scanner.next(results); 227 if (!results.isEmpty()) { 228 ++scannedRows; 229 } 230 } while (hasMoreRows); 231 } 232 233 assertEquals(rows.size(), scannedRows); 234 } finally { 235 StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT); 236 HBaseTestingUtil.closeRegionAndWAL(this.region); 237 } 238 } 239 240 @Test 241 public void testInitialLazySeekForUnfilteredExplicitColumnScan() throws Exception { 242 Scan scan = new Scan().withStartRow(LAZY_SEEK_ROW); 243 scan.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER); 244 assertInitialLazySeek(scan, true, true); 245 } 246 247 @Test 248 public void testInitialLazySeekForFilteredGet() throws Exception { 249 Get get = new Get(LAZY_SEEK_ROW); 250 get.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER); 251 get.setFilter(new PrefixFilter(LAZY_SEEK_ROW)); 252 assertInitialLazySeek(new Scan(get), true, true); 253 } 254 255 @Test 256 public void testInitialLazySeekForFilteredNonGetScan() throws Exception { 257 Scan scan = new Scan().withStartRow(LAZY_SEEK_ROW); 258 scan.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER); 259 scan.setFilter(new PrefixFilter(LAZY_SEEK_ROW)); 260 assertInitialLazySeek(scan, true, false); 261 } 262 263 @Test 264 public void testInitialLazySeekForAllColumnScan() throws Exception { 265 assertInitialLazySeek(new Scan().withStartRow(LAZY_SEEK_ROW), true, false); 266 } 267 268 @Test 269 public void testInitialLazySeekWhenDisabledGlobally() throws Exception { 270 Scan scan = new Scan().withStartRow(LAZY_SEEK_ROW); 271 scan.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER); 272 assertInitialLazySeek(scan, false, false); 273 } 274 275 private void assertInitialLazySeek(Scan scan, boolean lazySeekEnabled, boolean expected) 276 throws IOException { 277 StoreScanner.enableLazySeekGlobally(lazySeekEnabled); 278 try { 279 HStore store = createLazySeekTestStore(); 280 try (RecordingStoreScanner scanner = 281 new RecordingStoreScanner(store, scan, scan.getFamilyMap().get(LAZY_SEEK_FAMILY))) { 282 assertEquals(expected, scanner.initialSeekWasLazy); 283 } 284 } finally { 285 StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT); 286 if (this.region != null) { 287 HBaseTestingUtil.closeRegionAndWAL(this.region); 288 this.region = null; 289 } 290 } 291 } 292 293 private HStore createLazySeekTestStore() throws IOException { 294 TableDescriptor tableDescriptor = TableDescriptorBuilder 295 .newBuilder(TableName.valueOf("testInitialLazySeek")) 296 .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(LAZY_SEEK_FAMILY).build()).build(); 297 this.region = TEST_UTIL.createLocalHRegion(tableDescriptor, null, null); 298 Put put = new Put(LAZY_SEEK_ROW); 299 put.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER, Bytes.toBytes("value")); 300 region.put(put); 301 region.flush(true); 302 return region.getStore(LAZY_SEEK_FAMILY); 303 } 304 305 /** 306 * Test basic stop row filter works. 307 */ 308 @Test 309 public void testStopRow() throws Exception { 310 byte[] startrow = Bytes.toBytes("bbb"); 311 byte[] stoprow = Bytes.toBytes("ccc"); 312 try { 313 this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null); 314 HTestConst.addContent(this.region, HConstants.CATALOG_FAMILY); 315 List<Cell> results = new ArrayList<>(); 316 // Do simple test of getting one row only first. 317 Scan scan = new Scan().withStartRow(Bytes.toBytes("abc")).withStopRow(Bytes.toBytes("abd")); 318 scan.addFamily(HConstants.CATALOG_FAMILY); 319 320 InternalScanner s = region.getScanner(scan); 321 int count = 0; 322 while (s.next(results)) { 323 count++; 324 } 325 s.close(); 326 assertEquals(0, count); 327 // Now do something a bit more imvolved. 328 scan = new Scan().withStartRow(startrow).withStopRow(stoprow); 329 scan.addFamily(HConstants.CATALOG_FAMILY); 330 331 s = region.getScanner(scan); 332 count = 0; 333 Cell kv = null; 334 results = new ArrayList<>(); 335 for (boolean first = true; s.next(results);) { 336 kv = results.get(0); 337 if (first) { 338 assertTrue(CellUtil.matchingRows(kv, startrow)); 339 first = false; 340 } 341 count++; 342 } 343 assertTrue(Bytes.BYTES_COMPARATOR.compare(stoprow, CellUtil.cloneRow(kv)) > 0); 344 // We got something back. 345 assertTrue(count > 10); 346 s.close(); 347 } finally { 348 HBaseTestingUtil.closeRegionAndWAL(this.region); 349 } 350 } 351 352 void rowPrefixFilter(Scan scan) throws IOException { 353 List<Cell> results = new ArrayList<>(); 354 scan.addFamily(HConstants.CATALOG_FAMILY); 355 InternalScanner s = region.getScanner(scan); 356 boolean hasMore = true; 357 while (hasMore) { 358 hasMore = s.next(results); 359 for (Cell kv : results) { 360 assertEquals((byte) 'a', CellUtil.cloneRow(kv)[0]); 361 assertEquals((byte) 'b', CellUtil.cloneRow(kv)[1]); 362 } 363 results.clear(); 364 } 365 s.close(); 366 } 367 368 void rowInclusiveStopFilter(Scan scan, byte[] stopRow) throws IOException { 369 List<Cell> results = new ArrayList<>(); 370 scan.addFamily(HConstants.CATALOG_FAMILY); 371 InternalScanner s = region.getScanner(scan); 372 boolean hasMore = true; 373 while (hasMore) { 374 hasMore = s.next(results); 375 for (Cell kv : results) { 376 assertTrue(Bytes.compareTo(CellUtil.cloneRow(kv), stopRow) <= 0); 377 } 378 results.clear(); 379 } 380 s.close(); 381 } 382 383 @Test 384 public void testFilters() throws IOException { 385 try { 386 this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null); 387 HTestConst.addContent(this.region, HConstants.CATALOG_FAMILY); 388 byte[] prefix = Bytes.toBytes("ab"); 389 Filter newFilter = new PrefixFilter(prefix); 390 Scan scan = new Scan(); 391 scan.setFilter(newFilter); 392 rowPrefixFilter(scan); 393 394 byte[] stopRow = Bytes.toBytes("bbc"); 395 newFilter = new WhileMatchFilter(new InclusiveStopFilter(stopRow)); 396 scan = new Scan(); 397 scan.setFilter(newFilter); 398 rowInclusiveStopFilter(scan, stopRow); 399 400 } finally { 401 HBaseTestingUtil.closeRegionAndWAL(this.region); 402 } 403 } 404 405 /** 406 * Test that closing a scanner while a client is using it doesn't throw NPEs but instead a 407 * UnknownScannerException. HBASE-2503 408 */ 409 @Test 410 public void testRaceBetweenClientAndTimeout() throws Exception { 411 try { 412 this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null); 413 HTestConst.addContent(this.region, HConstants.CATALOG_FAMILY); 414 Scan scan = new Scan(); 415 InternalScanner s = region.getScanner(scan); 416 List<Cell> results = new ArrayList<>(); 417 try { 418 s.next(results); 419 s.close(); 420 s.next(results); 421 fail("We don't want anything more, we should be failing"); 422 } catch (UnknownScannerException ex) { 423 // ok! 424 return; 425 } 426 } finally { 427 HBaseTestingUtil.closeRegionAndWAL(this.region); 428 } 429 } 430 431 /** 432 * The test! 433 */ 434 @Test 435 public void testScanner() throws IOException { 436 try { 437 region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null); 438 Table table = new RegionAsTable(region); 439 440 // Write information to the meta table 441 442 Put put = new Put(ROW_KEY, EnvironmentEdgeManager.currentTime()); 443 444 put.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER, 445 RegionInfo.toByteArray(REGION_INFO)); 446 table.put(put); 447 448 // What we just committed is in the memstore. Verify that we can get 449 // it back both with scanning and get 450 451 scan(false, null); 452 getRegionInfo(table); 453 454 // Close and re-open 455 456 ((HRegion) region).close(); 457 region = HRegion.openHRegion(region, null); 458 table = new RegionAsTable(region); 459 460 // Verify we can get the data back now that it is on disk. 461 462 scan(false, null); 463 getRegionInfo(table); 464 465 // Store some new information 466 467 String address = HConstants.LOCALHOST_IP + ":" + HBaseTestingUtil.randomFreePort(); 468 469 put = new Put(ROW_KEY, EnvironmentEdgeManager.currentTime()); 470 put.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER, Bytes.toBytes(address)); 471 472 // put.add(HConstants.COL_STARTCODE, Bytes.toBytes(START_CODE)); 473 474 table.put(put); 475 476 // Validate that we can still get the HRegionInfo, even though it is in 477 // an older row on disk and there is a newer row in the memstore 478 479 scan(true, address.toString()); 480 getRegionInfo(table); 481 482 // flush cache 483 this.region.flush(true); 484 485 // Validate again 486 487 scan(true, address.toString()); 488 getRegionInfo(table); 489 490 // Close and reopen 491 492 ((HRegion) region).close(); 493 region = HRegion.openHRegion(region, null); 494 table = new RegionAsTable(region); 495 496 // Validate again 497 498 scan(true, address.toString()); 499 getRegionInfo(table); 500 501 // Now update the information again 502 503 address = "bar.foo.com:4321"; 504 505 put = new Put(ROW_KEY, EnvironmentEdgeManager.currentTime()); 506 507 put.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER, Bytes.toBytes(address)); 508 table.put(put); 509 510 // Validate again 511 512 scan(true, address.toString()); 513 getRegionInfo(table); 514 515 // flush cache 516 517 region.flush(true); 518 519 // Validate again 520 521 scan(true, address.toString()); 522 getRegionInfo(table); 523 524 // Close and reopen 525 526 ((HRegion) this.region).close(); 527 this.region = HRegion.openHRegion(region, null); 528 table = new RegionAsTable(this.region); 529 530 // Validate again 531 532 scan(true, address.toString()); 533 getRegionInfo(table); 534 535 } finally { 536 // clean up 537 HBaseTestingUtil.closeRegionAndWAL(this.region); 538 } 539 } 540 541 /** Compare the HRegionInfo we read from HBase to what we stored */ 542 private void validateRegionInfo(byte[] regionBytes) throws IOException { 543 RegionInfo info = RegionInfo.parseFromOrNull(regionBytes); 544 545 assertEquals(REGION_INFO.getRegionId(), info.getRegionId()); 546 assertEquals(0, info.getStartKey().length); 547 assertEquals(0, info.getEndKey().length); 548 assertEquals(0, Bytes.compareTo(info.getRegionName(), REGION_INFO.getRegionName())); 549 // assertEquals(0, info.getTableDesc().compareTo(REGION_INFO.getTableDesc())); 550 } 551 552 /** Use a scanner to get the region info and then validate the results */ 553 private void scan(boolean validateStartcode, String serverName) throws IOException { 554 InternalScanner scanner = null; 555 Scan scan = null; 556 List<Cell> results = new ArrayList<>(); 557 byte[][][] scanColumns = { COLS, EXPLICIT_COLS }; 558 for (int i = 0; i < scanColumns.length; i++) { 559 try { 560 scan = new Scan().withStartRow(FIRST_ROW); 561 for (int ii = 0; ii < EXPLICIT_COLS.length; ii++) { 562 scan.addColumn(COLS[0], EXPLICIT_COLS[ii]); 563 } 564 scanner = region.getScanner(scan); 565 while (scanner.next(results)) { 566 assertTrue( 567 hasColumn(results, HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER)); 568 byte[] val = CellUtil.cloneValue( 569 getColumn(results, HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER)); 570 validateRegionInfo(val); 571 if (validateStartcode) { 572 // assertTrue(hasColumn(results, HConstants.CATALOG_FAMILY, 573 // HConstants.STARTCODE_QUALIFIER)); 574 // val = getColumn(results, HConstants.CATALOG_FAMILY, 575 // HConstants.STARTCODE_QUALIFIER).getValue(); 576 assertNotNull(val); 577 assertFalse(val.length == 0); 578 long startCode = Bytes.toLong(val); 579 assertEquals(START_CODE, startCode); 580 } 581 582 if (serverName != null) { 583 assertTrue(hasColumn(results, HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER)); 584 val = CellUtil.cloneValue( 585 getColumn(results, HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER)); 586 assertNotNull(val); 587 assertFalse(val.length == 0); 588 String server = Bytes.toString(val); 589 assertEquals(0, server.compareTo(serverName)); 590 } 591 } 592 } finally { 593 InternalScanner s = scanner; 594 scanner = null; 595 if (s != null) { 596 s.close(); 597 } 598 } 599 } 600 } 601 602 private boolean hasColumn(final List<Cell> kvs, final byte[] family, final byte[] qualifier) { 603 for (Cell kv : kvs) { 604 if (CellUtil.matchingFamily(kv, family) && CellUtil.matchingQualifier(kv, qualifier)) { 605 return true; 606 } 607 } 608 return false; 609 } 610 611 private Cell getColumn(final List<Cell> kvs, final byte[] family, final byte[] qualifier) { 612 for (Cell kv : kvs) { 613 if (CellUtil.matchingFamily(kv, family) && CellUtil.matchingQualifier(kv, qualifier)) { 614 return kv; 615 } 616 } 617 return null; 618 } 619 620 /** Use get to retrieve the HRegionInfo and validate it */ 621 private void getRegionInfo(Table table) throws IOException { 622 Get get = new Get(ROW_KEY); 623 get.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER); 624 Result result = table.get(get); 625 byte[] bytes = result.value(); 626 validateRegionInfo(bytes); 627 } 628 629 /** 630 * Tests to do a sync flush during the middle of a scan. This is testing the StoreScanner update 631 * readers code essentially. This is not highly concurrent, since its all 1 thread. HBase-910. 632 */ 633 @Test 634 public void testScanAndSyncFlush() throws Exception { 635 this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null); 636 Table hri = new RegionAsTable(region); 637 try { 638 LOG.info("Added: " + HTestConst.addContent(hri, Bytes.toString(HConstants.CATALOG_FAMILY), 639 Bytes.toString(HConstants.REGIONINFO_QUALIFIER))); 640 int count = count(hri, -1, false); 641 assertEquals(count, count(hri, 100, false)); // do a sync flush. 642 } catch (Exception e) { 643 LOG.error("Failed", e); 644 throw e; 645 } finally { 646 HBaseTestingUtil.closeRegionAndWAL(this.region); 647 } 648 } 649 650 /** 651 * Tests to do a concurrent flush (using a 2nd thread) while scanning. This tests both the 652 * StoreScanner update readers and the transition from memstore -> snapshot -> store file. 653 */ 654 @Test 655 public void testScanAndRealConcurrentFlush() throws Exception { 656 this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null); 657 Table hri = new RegionAsTable(region); 658 try { 659 LOG.info("Added: " + HTestConst.addContent(hri, Bytes.toString(HConstants.CATALOG_FAMILY), 660 Bytes.toString(HConstants.REGIONINFO_QUALIFIER))); 661 int count = count(hri, -1, false); 662 assertEquals(count, count(hri, 100, true)); // do a true concurrent background thread flush 663 } catch (Exception e) { 664 LOG.error("Failed", e); 665 throw e; 666 } finally { 667 HBaseTestingUtil.closeRegionAndWAL(this.region); 668 } 669 } 670 671 /** 672 * Make sure scanner returns correct result when we run a major compaction with deletes. 673 */ 674 @Test 675 public void testScanAndConcurrentMajorCompact(TestInfo testInfo) throws Exception { 676 TableDescriptor htd = 677 TEST_UTIL.createTableDescriptor(TableName.valueOf(testInfo.getTestMethod().get().getName()), 678 ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3, HConstants.FOREVER, 679 ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED); 680 this.region = TEST_UTIL.createLocalHRegion(htd, null, null); 681 Table hri = new RegionAsTable(region); 682 683 try { 684 HTestConst.addContent(hri, Bytes.toString(fam1), Bytes.toString(col1), firstRowBytes, 685 secondRowBytes); 686 HTestConst.addContent(hri, Bytes.toString(fam2), Bytes.toString(col1), firstRowBytes, 687 secondRowBytes); 688 689 Delete dc = new Delete(firstRowBytes); 690 /* delete column1 of firstRow */ 691 dc.addColumns(fam1, col1); 692 region.delete(dc); 693 region.flush(true); 694 695 HTestConst.addContent(hri, Bytes.toString(fam1), Bytes.toString(col1), secondRowBytes, 696 thirdRowBytes); 697 HTestConst.addContent(hri, Bytes.toString(fam2), Bytes.toString(col1), secondRowBytes, 698 thirdRowBytes); 699 region.flush(true); 700 701 InternalScanner s = region.getScanner(new Scan()); 702 // run a major compact, column1 of firstRow will be cleaned. 703 region.compact(true); 704 705 List<Cell> results = new ArrayList<>(); 706 s.next(results); 707 708 // make sure returns column2 of firstRow 709 assertEquals(1, results.size(), "result is not correct, keyValues : " + results); 710 assertTrue(CellUtil.matchingRows(results.get(0), firstRowBytes)); 711 assertTrue(CellUtil.matchingFamily(results.get(0), fam2)); 712 713 results = new ArrayList<>(); 714 s.next(results); 715 716 // get secondRow 717 assertEquals(2, results.size()); 718 assertTrue(CellUtil.matchingRows(results.get(0), secondRowBytes)); 719 assertTrue(CellUtil.matchingFamily(results.get(0), fam1)); 720 assertTrue(CellUtil.matchingFamily(results.get(1), fam2)); 721 } finally { 722 HBaseTestingUtil.closeRegionAndWAL(this.region); 723 } 724 } 725 726 /** 727 * Count table. 728 * @param countTable Table 729 * @param flushIndex At what row we start the flush. 730 * @param concurrent if the flush should be concurrent or sync. 731 * @return Count of rows found. 732 */ 733 private int count(final Table countTable, final int flushIndex, boolean concurrent) 734 throws Exception { 735 LOG.info("Taking out counting scan"); 736 Scan scan = new Scan(); 737 for (byte[] qualifier : EXPLICIT_COLS) { 738 scan.addColumn(HConstants.CATALOG_FAMILY, qualifier); 739 } 740 ResultScanner s = countTable.getScanner(scan); 741 int count = 0; 742 boolean justFlushed = false; 743 while (s.next() != null) { 744 if (justFlushed) { 745 LOG.info("after next() just after next flush"); 746 justFlushed = false; 747 } 748 count++; 749 if (flushIndex == count) { 750 LOG.info("Starting flush at flush index " + flushIndex); 751 Thread t = new Thread() { 752 @Override 753 public void run() { 754 try { 755 region.flush(true); 756 LOG.info("Finishing flush"); 757 } catch (IOException e) { 758 LOG.info("Failed flush cache"); 759 } 760 } 761 }; 762 t.start(); 763 if (!concurrent) { 764 // sync flush 765 t.join(); 766 } 767 LOG.info("Continuing on after kicking off background flush"); 768 justFlushed = true; 769 } 770 } 771 s.close(); 772 LOG.info("Found " + count + " items"); 773 return count; 774 } 775}