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 java.io.IOException; 021import java.io.InterruptedIOException; 022import java.util.ArrayList; 023import java.util.List; 024import java.util.NavigableSet; 025import java.util.Optional; 026import java.util.concurrent.CountDownLatch; 027import java.util.concurrent.locks.ReentrantLock; 028import org.apache.hadoop.hbase.Cell; 029import org.apache.hadoop.hbase.CellComparator; 030import org.apache.hadoop.hbase.CellUtil; 031import org.apache.hadoop.hbase.DoNotRetryIOException; 032import org.apache.hadoop.hbase.ExtendedCell; 033import org.apache.hadoop.hbase.HConstants; 034import org.apache.hadoop.hbase.KeyValue; 035import org.apache.hadoop.hbase.KeyValueUtil; 036import org.apache.hadoop.hbase.PrivateCellUtil; 037import org.apache.hadoop.hbase.PrivateConstants; 038import org.apache.hadoop.hbase.client.IsolationLevel; 039import org.apache.hadoop.hbase.client.Scan; 040import org.apache.hadoop.hbase.executor.ExecutorService; 041import org.apache.hadoop.hbase.filter.Filter; 042import org.apache.hadoop.hbase.ipc.RpcCall; 043import org.apache.hadoop.hbase.ipc.RpcServer; 044import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope; 045import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState; 046import org.apache.hadoop.hbase.regionserver.handler.ParallelSeekHandler; 047import org.apache.hadoop.hbase.regionserver.querymatcher.CompactionScanQueryMatcher; 048import org.apache.hadoop.hbase.regionserver.querymatcher.ScanQueryMatcher; 049import org.apache.hadoop.hbase.regionserver.querymatcher.UserScanQueryMatcher; 050import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 051import org.apache.yetus.audience.InterfaceAudience; 052import org.slf4j.Logger; 053import org.slf4j.LoggerFactory; 054 055import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; 056import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils; 057 058/** 059 * Scanner scans both the memstore and the Store. Coalesce KeyValue stream into List<KeyValue> 060 * for a single row. 061 * <p> 062 * The implementation is not thread safe. So there will be no race between next and close. The only 063 * exception is updateReaders, it will be called in the memstore flush thread to indicate that there 064 * is a flush. 065 */ 066@InterfaceAudience.Private 067public class StoreScanner extends NonReversedNonLazyKeyValueScanner 068 implements KeyValueScanner, InternalScanner, ChangedReadersObserver { 069 private static final Logger LOG = LoggerFactory.getLogger(StoreScanner.class); 070 // In unit tests, the store could be null 071 protected final HStore store; 072 private final CellComparator comparator; 073 private ScanQueryMatcher matcher; 074 protected KeyValueHeap heap; 075 private boolean cacheBlocks; 076 077 private long countPerRow = 0; 078 private int storeLimit = -1; 079 private int storeOffset = 0; 080 081 // Used to indicate that the scanner has closed (see HBASE-1107) 082 private volatile boolean closing = false; 083 private final boolean get; 084 private final boolean explicitColumnQuery; 085 private final boolean useRowColBloom; 086 /** 087 * A flag that enables StoreFileScanner parallel-seeking 088 */ 089 private boolean parallelSeekEnabled = false; 090 private ExecutorService executor; 091 private final Scan scan; 092 private final long oldestUnexpiredTS; 093 private final long now; 094 private final int minVersions; 095 private final long maxRowSize; 096 private final long cellsPerHeartbeatCheck; 097 long memstoreOnlyReads; 098 long mixedReads; 099 100 // 1) Collects all the KVHeap that are eagerly getting closed during the 101 // course of a scan 102 // 2) Collects the unused memstore scanners. If we close the memstore scanners 103 // before sending data to client, the chunk may be reclaimed by other 104 // updates and the data will be corrupt. 105 private final List<KeyValueScanner> scannersForDelayedClose = new ArrayList<>(); 106 107 /** 108 * The number of KVs seen by the scanner. Includes explicitly skipped KVs, but not KVs skipped via 109 * seeking to next row/column. TODO: estimate them? 110 */ 111 private long kvsScanned = 0; 112 private ExtendedCell prevCell = null; 113 114 private final long preadMaxBytes; 115 private long bytesRead; 116 117 /** We don't ever expect to change this, the constant is just for clarity. */ 118 static final boolean LAZY_SEEK_ENABLED_BY_DEFAULT = true; 119 public static final String STORESCANNER_PARALLEL_SEEK_ENABLE = 120 "hbase.storescanner.parallel.seek.enable"; 121 122 /** Used during unit testing to ensure that lazy seek does save seek ops */ 123 private static boolean lazySeekEnabledGlobally = LAZY_SEEK_ENABLED_BY_DEFAULT; 124 125 /** 126 * The number of cells scanned in between timeout checks. Specifying a larger value means that 127 * timeout checks will occur less frequently. Specifying a small value will lead to more frequent 128 * timeout checks. 129 */ 130 public static final String HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK = 131 "hbase.cells.scanned.per.heartbeat.check"; 132 133 /** 134 * Default value of {@link #HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK}. 135 */ 136 public static final long DEFAULT_HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK = 10000; 137 138 /** 139 * If the read type is Scan.ReadType.DEFAULT, we will start with pread, and if the kvs we scanned 140 * reaches this limit, we will reopen the scanner with stream. The default value is 4 times of 141 * block size for this store. If configured with a value <0, for all scans with ReadType DEFAULT, 142 * we will open scanner with stream mode itself. 143 */ 144 public static final String STORESCANNER_PREAD_MAX_BYTES = "hbase.storescanner.pread.max.bytes"; 145 146 private final Scan.ReadType readType; 147 148 // A flag whether use pread for scan 149 // it maybe changed if we use Scan.ReadType.DEFAULT and we have read lots of data. 150 private boolean scanUsePread; 151 // Indicates whether there was flush during the course of the scan 152 private volatile boolean flushed = false; 153 // generally we get one file from a flush 154 private final List<KeyValueScanner> flushedstoreFileScanners = new ArrayList<>(1); 155 // Since CompactingMemstore is now default, we get three memstore scanners from a flush 156 private final List<KeyValueScanner> memStoreScannersAfterFlush = new ArrayList<>(3); 157 // The current list of scanners 158 final List<KeyValueScanner> currentScanners = new ArrayList<>(); 159 // flush update lock 160 private final ReentrantLock flushLock = new ReentrantLock(); 161 // lock for closing. 162 private final ReentrantLock closeLock = new ReentrantLock(); 163 164 protected final long readPt; 165 private boolean topChanged = false; 166 167 /** An internal constructor. */ 168 private StoreScanner(HStore store, Scan scan, ScanInfo scanInfo, int numColumns, long readPt, 169 boolean cacheBlocks, ScanType scanType) { 170 this.readPt = readPt; 171 this.store = store; 172 this.cacheBlocks = cacheBlocks; 173 this.comparator = Preconditions.checkNotNull(scanInfo.getComparator()); 174 get = scan.isGetScan(); 175 explicitColumnQuery = numColumns > 0; 176 this.scan = scan; 177 this.now = EnvironmentEdgeManager.currentTime(); 178 this.oldestUnexpiredTS = scan.isRaw() ? 0L : now - scanInfo.getTtl(); 179 this.minVersions = scanInfo.getMinVersions(); 180 181 // We look up row-column Bloom filters for multi-column queries as part of 182 // the seek operation. However, we also look the row-column Bloom filter 183 // for multi-row (non-"get") scans because this is not done in 184 // StoreFile.passesBloomFilter(Scan, SortedSet<byte[]>). 185 this.useRowColBloom = numColumns > 1 || (!get && numColumns == 1) && (store == null 186 || store.getColumnFamilyDescriptor().getBloomFilterType() == BloomType.ROWCOL); 187 this.maxRowSize = scanInfo.getTableMaxRowSize(); 188 this.preadMaxBytes = scanInfo.getPreadMaxBytes(); 189 if (get) { 190 this.readType = Scan.ReadType.PREAD; 191 this.scanUsePread = true; 192 } else if (scanType != ScanType.USER_SCAN) { 193 // For compaction scanners never use Pread as already we have stream based scanners on the 194 // store files to be compacted 195 this.readType = Scan.ReadType.STREAM; 196 this.scanUsePread = false; 197 } else { 198 if (scan.getReadType() == Scan.ReadType.DEFAULT) { 199 if (scanInfo.isUsePread()) { 200 this.readType = Scan.ReadType.PREAD; 201 } else if (this.preadMaxBytes < 0) { 202 this.readType = Scan.ReadType.STREAM; 203 } else { 204 this.readType = Scan.ReadType.DEFAULT; 205 } 206 } else { 207 this.readType = scan.getReadType(); 208 } 209 // Always start with pread unless user specific stream. Will change to stream later if 210 // readType is default if the scan keeps running for a long time. 211 this.scanUsePread = this.readType != Scan.ReadType.STREAM; 212 } 213 this.cellsPerHeartbeatCheck = scanInfo.getCellsPerTimeoutCheck(); 214 // Parallel seeking is on if the config allows and more there is more than one store file. 215 if (store != null && store.getStorefilesCount() > 1) { 216 RegionServerServices rsService = store.getHRegion().getRegionServerServices(); 217 if (rsService != null && scanInfo.isParallelSeekEnabled()) { 218 this.parallelSeekEnabled = true; 219 this.executor = rsService.getExecutorService(); 220 } 221 } 222 } 223 224 private void addCurrentScanners(List<? extends KeyValueScanner> scanners) { 225 this.currentScanners.addAll(scanners); 226 } 227 228 private static boolean isOnlyLatestVersionScan(Scan scan) { 229 // No need to check for Scan#getMaxVersions because live version files generated by store file 230 // writer retains max versions specified in ColumnFamilyDescriptor for the given CF 231 return !scan.isRaw() && scan.getTimeRange().getMax() == HConstants.LATEST_TIMESTAMP; 232 } 233 234 /** 235 * Opens a scanner across memstore, snapshot, and all StoreFiles. Assumes we are not in a 236 * compaction. 237 * @param store who we scan 238 * @param scan the spec 239 * @param columns which columns we are scanning 240 */ 241 public StoreScanner(HStore store, ScanInfo scanInfo, Scan scan, NavigableSet<byte[]> columns, 242 long readPt) throws IOException { 243 this(store, scan, scanInfo, columns != null ? columns.size() : 0, readPt, scan.getCacheBlocks(), 244 ScanType.USER_SCAN); 245 if (columns != null && scan.isRaw()) { 246 throw new DoNotRetryIOException("Cannot specify any column for a raw scan"); 247 } 248 matcher = UserScanQueryMatcher.create(scan, scanInfo, columns, oldestUnexpiredTS, now, 249 store.getCoprocessorHost()); 250 251 store.addChangedReaderObserver(this); 252 253 List<KeyValueScanner> scanners = null; 254 try { 255 // Pass columns to try to filter out unnecessary StoreFiles. 256 scanners = selectScannersFrom(store, 257 store.getScanners(cacheBlocks, scanUsePread, false, matcher, scan.getStartRow(), 258 scan.includeStartRow(), scan.getStopRow(), scan.includeStopRow(), this.readPt, 259 isOnlyLatestVersionScan(scan))); 260 261 // Seek all scanners to the start of the Row (or if the exact matching row 262 // key does not exist, then to the start of the next matching Row). 263 // Always check bloom filter to optimize the top row seek for delete 264 // family marker. 265 seekScanners(scanners, matcher.getStartKey(), explicitColumnQuery && lazySeekEnabledGlobally, 266 parallelSeekEnabled); 267 268 // set storeLimit 269 this.storeLimit = scan.getMaxResultsPerColumnFamily(); 270 271 // set rowOffset 272 this.storeOffset = scan.getRowOffsetPerColumnFamily(); 273 addCurrentScanners(scanners); 274 // Combine all seeked scanners with a heap 275 resetKVHeap(scanners, comparator); 276 } catch (IOException e) { 277 clearAndClose(scanners); 278 // remove us from the HStore#changedReaderObservers here or we'll have no chance to 279 // and might cause memory leak 280 store.deleteChangedReaderObserver(this); 281 throw e; 282 } 283 } 284 285 // a dummy scan instance for compaction. 286 private static final Scan SCAN_FOR_COMPACTION = new Scan(); 287 288 /** 289 * Used for store file compaction and memstore compaction. 290 * <p> 291 * Opens a scanner across specified StoreFiles/MemStoreSegments. 292 * @param store who we scan 293 * @param scanners ancillary scanners 294 * @param smallestReadPoint the readPoint that we should use for tracking versions 295 */ 296 public StoreScanner(HStore store, ScanInfo scanInfo, List<? extends KeyValueScanner> scanners, 297 ScanType scanType, long smallestReadPoint, long earliestPutTs) throws IOException { 298 this(store, scanInfo, scanners, scanType, smallestReadPoint, earliestPutTs, null, null); 299 } 300 301 /** 302 * Used for compactions that drop deletes from a limited range of rows. 303 * <p> 304 * Opens a scanner across specified StoreFiles. 305 * @param store who we scan 306 * @param scanners ancillary scanners 307 * @param smallestReadPoint the readPoint that we should use for tracking versions 308 * @param dropDeletesFromRow The inclusive left bound of the range; can be EMPTY_START_ROW. 309 * @param dropDeletesToRow The exclusive right bound of the range; can be EMPTY_END_ROW. 310 */ 311 public StoreScanner(HStore store, ScanInfo scanInfo, List<? extends KeyValueScanner> scanners, 312 long smallestReadPoint, long earliestPutTs, byte[] dropDeletesFromRow, byte[] dropDeletesToRow) 313 throws IOException { 314 this(store, scanInfo, scanners, ScanType.COMPACT_RETAIN_DELETES, smallestReadPoint, 315 earliestPutTs, dropDeletesFromRow, dropDeletesToRow); 316 } 317 318 private StoreScanner(HStore store, ScanInfo scanInfo, List<? extends KeyValueScanner> scanners, 319 ScanType scanType, long smallestReadPoint, long earliestPutTs, byte[] dropDeletesFromRow, 320 byte[] dropDeletesToRow) throws IOException { 321 this(store, SCAN_FOR_COMPACTION, scanInfo, 0, 322 store.getHRegion().getReadPoint(IsolationLevel.READ_COMMITTED), false, scanType); 323 assert scanType != ScanType.USER_SCAN; 324 matcher = 325 CompactionScanQueryMatcher.create(scanInfo, scanType, smallestReadPoint, earliestPutTs, 326 oldestUnexpiredTS, now, dropDeletesFromRow, dropDeletesToRow, store.getCoprocessorHost()); 327 328 // Filter the list of scanners using Bloom filters, time range, TTL, etc. 329 scanners = selectScannersFrom(store, scanners); 330 331 // Seek all scanners to the initial key 332 seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled); 333 addCurrentScanners(scanners); 334 // Combine all seeked scanners with a heap 335 resetKVHeap(scanners, comparator); 336 } 337 338 private void seekAllScanner(ScanInfo scanInfo, List<? extends KeyValueScanner> scanners) 339 throws IOException { 340 // Seek all scanners to the initial key 341 seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled); 342 addCurrentScanners(scanners); 343 resetKVHeap(scanners, comparator); 344 } 345 346 // For mob compaction only as we do not have a Store instance when doing mob compaction. 347 public StoreScanner(ScanInfo scanInfo, ScanType scanType, 348 List<? extends KeyValueScanner> scanners) throws IOException { 349 this(null, SCAN_FOR_COMPACTION, scanInfo, 0, Long.MAX_VALUE, false, scanType); 350 assert scanType != ScanType.USER_SCAN; 351 this.matcher = CompactionScanQueryMatcher.create(scanInfo, scanType, Long.MAX_VALUE, 0L, 352 oldestUnexpiredTS, now, null, null, null); 353 seekAllScanner(scanInfo, scanners); 354 } 355 356 // Used to instantiate a scanner for user scan in test 357 StoreScanner(Scan scan, ScanInfo scanInfo, NavigableSet<byte[]> columns, 358 List<? extends KeyValueScanner> scanners, ScanType scanType) throws IOException { 359 // 0 is passed as readpoint because the test bypasses Store 360 this(null, scan, scanInfo, columns != null ? columns.size() : 0, 0L, scan.getCacheBlocks(), 361 scanType); 362 if (scanType == ScanType.USER_SCAN) { 363 this.matcher = 364 UserScanQueryMatcher.create(scan, scanInfo, columns, oldestUnexpiredTS, now, null); 365 } else { 366 this.matcher = CompactionScanQueryMatcher.create(scanInfo, scanType, Long.MAX_VALUE, 367 PrivateConstants.OLDEST_TIMESTAMP, oldestUnexpiredTS, now, null, null, null); 368 } 369 seekAllScanner(scanInfo, scanners); 370 } 371 372 // Used to instantiate a scanner for user scan in test 373 StoreScanner(Scan scan, ScanInfo scanInfo, NavigableSet<byte[]> columns, 374 List<? extends KeyValueScanner> scanners) throws IOException { 375 // 0 is passed as readpoint because the test bypasses Store 376 this(null, scan, scanInfo, columns != null ? columns.size() : 0, 0L, scan.getCacheBlocks(), 377 ScanType.USER_SCAN); 378 this.matcher = 379 UserScanQueryMatcher.create(scan, scanInfo, columns, oldestUnexpiredTS, now, null); 380 seekAllScanner(scanInfo, scanners); 381 } 382 383 // Used to instantiate a scanner for compaction in test 384 StoreScanner(ScanInfo scanInfo, int maxVersions, ScanType scanType, 385 List<? extends KeyValueScanner> scanners) throws IOException { 386 // 0 is passed as readpoint because the test bypasses Store 387 this(null, maxVersions > 0 ? new Scan().readVersions(maxVersions) : SCAN_FOR_COMPACTION, 388 scanInfo, 0, 0L, false, scanType); 389 this.matcher = CompactionScanQueryMatcher.create(scanInfo, scanType, Long.MAX_VALUE, 390 PrivateConstants.OLDEST_TIMESTAMP, oldestUnexpiredTS, now, null, null, null); 391 seekAllScanner(scanInfo, scanners); 392 } 393 394 boolean isScanUsePread() { 395 return this.scanUsePread; 396 } 397 398 /** 399 * Seek the specified scanners with the given key 400 * @param isLazy true if using lazy seek 401 * @param isParallelSeek true if using parallel seek 402 */ 403 protected void seekScanners(List<? extends KeyValueScanner> scanners, ExtendedCell seekKey, 404 boolean isLazy, boolean isParallelSeek) throws IOException { 405 // Seek all scanners to the start of the Row (or if the exact matching row 406 // key does not exist, then to the start of the next matching Row). 407 // Always check bloom filter to optimize the top row seek for delete 408 // family marker. 409 if (isLazy) { 410 for (KeyValueScanner scanner : scanners) { 411 scanner.requestSeek(seekKey, false, true); 412 } 413 } else { 414 if (!isParallelSeek) { 415 long totalScannersSoughtBytes = 0; 416 for (KeyValueScanner scanner : scanners) { 417 if (matcher.isUserScan() && totalScannersSoughtBytes >= maxRowSize) { 418 throw new RowTooBigException( 419 "Max row size allowed: " + maxRowSize + ", but row is bigger than that"); 420 } 421 scanner.seek(seekKey); 422 Cell c = scanner.peek(); 423 if (c != null) { 424 totalScannersSoughtBytes += PrivateCellUtil.estimatedSerializedSizeOf(c); 425 } 426 } 427 } else { 428 parallelSeek(scanners, seekKey); 429 } 430 } 431 } 432 433 protected void resetKVHeap(List<? extends KeyValueScanner> scanners, CellComparator comparator) 434 throws IOException { 435 // Combine all seeked scanners with a heap 436 heap = newKVHeap(scanners, comparator); 437 } 438 439 protected KeyValueHeap newKVHeap(List<? extends KeyValueScanner> scanners, 440 CellComparator comparator) throws IOException { 441 return new KeyValueHeap(scanners, comparator); 442 } 443 444 /** 445 * Filters the given list of scanners using Bloom filter, time range, and TTL. 446 * <p> 447 * Will be overridden by testcase so declared as protected. 448 */ 449 protected List<KeyValueScanner> selectScannersFrom(HStore store, 450 List<? extends KeyValueScanner> allScanners) { 451 boolean memOnly; 452 boolean filesOnly; 453 if (scan instanceof InternalScan) { 454 InternalScan iscan = (InternalScan) scan; 455 memOnly = iscan.isCheckOnlyMemStore(); 456 filesOnly = iscan.isCheckOnlyStoreFiles(); 457 } else { 458 memOnly = false; 459 filesOnly = false; 460 } 461 462 List<KeyValueScanner> scanners = new ArrayList<>(allScanners.size()); 463 464 // We can only exclude store files based on TTL if minVersions is set to 0. 465 // Otherwise, we might have to return KVs that have technically expired. 466 long expiredTimestampCutoff = minVersions == 0 ? oldestUnexpiredTS : Long.MIN_VALUE; 467 468 // include only those scan files which pass all filters 469 for (KeyValueScanner kvs : allScanners) { 470 boolean isFile = kvs.isFileScanner(); 471 if ((!isFile && filesOnly) || (isFile && memOnly)) { 472 kvs.close(); 473 continue; 474 } 475 476 if (kvs.shouldUseScanner(scan, store, expiredTimestampCutoff)) { 477 scanners.add(kvs); 478 } else { 479 kvs.close(); 480 } 481 } 482 return scanners; 483 } 484 485 @Override 486 public ExtendedCell peek() { 487 return heap != null ? heap.peek() : null; 488 } 489 490 @Override 491 public KeyValue next() { 492 // throw runtime exception perhaps? 493 throw new RuntimeException("Never call StoreScanner.next()"); 494 } 495 496 @Override 497 public void close() { 498 close(true); 499 } 500 501 private void close(boolean withDelayedScannersClose) { 502 closeLock.lock(); 503 // If the closeLock is acquired then any subsequent updateReaders() 504 // call is ignored. 505 try { 506 if (this.closing) { 507 return; 508 } 509 if (withDelayedScannersClose) { 510 this.closing = true; 511 } 512 // For mob compaction, we do not have a store. 513 if (this.store != null) { 514 this.store.deleteChangedReaderObserver(this); 515 } 516 if (withDelayedScannersClose) { 517 clearAndClose(scannersForDelayedClose); 518 clearAndClose(memStoreScannersAfterFlush); 519 clearAndClose(flushedstoreFileScanners); 520 if (this.heap != null) { 521 this.heap.close(); 522 this.currentScanners.clear(); 523 this.heap = null; // CLOSED! 524 } 525 } else { 526 if (this.heap != null) { 527 this.scannersForDelayedClose.add(this.heap); 528 this.currentScanners.clear(); 529 this.heap = null; 530 } 531 } 532 } finally { 533 closeLock.unlock(); 534 } 535 } 536 537 @Override 538 public boolean seek(ExtendedCell key) throws IOException { 539 if (checkFlushed()) { 540 reopenAfterFlush(); 541 } 542 return this.heap.seek(key); 543 } 544 545 /** 546 * Get the next row of values from this Store. 547 * @return true if there are more rows, false if scanner is done 548 */ 549 @Override 550 public boolean next(List<Cell> outResult, ScannerContext scannerContext) throws IOException { 551 if (scannerContext == null) { 552 throw new IllegalArgumentException("Scanner context cannot be null"); 553 } 554 if (checkFlushed() && reopenAfterFlush()) { 555 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues(); 556 } 557 558 // if the heap was left null, then the scanners had previously run out anyways, close and 559 // return. 560 if (this.heap == null) { 561 // By this time partial close should happened because already heap is null 562 close(false);// Do all cleanup except heap.close() 563 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 564 } 565 566 ExtendedCell cell = this.heap.peek(); 567 if (cell == null) { 568 close(false);// Do all cleanup except heap.close() 569 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 570 } 571 572 // only call setRow if the row changes; avoids confusing the query matcher 573 // if scanning intra-row 574 575 // If no limits exists in the scope LimitScope.Between_Cells then we are sure we are changing 576 // rows. Else it is possible we are still traversing the same row so we must perform the row 577 // comparison. 578 if (!scannerContext.hasAnyLimit(LimitScope.BETWEEN_CELLS) || matcher.currentRow() == null) { 579 this.countPerRow = 0; 580 matcher.setToNewRow(cell); 581 } 582 583 // Clear progress away unless invoker has indicated it should be kept. 584 if (!scannerContext.getKeepProgress() && !scannerContext.getSkippingRow()) { 585 scannerContext.clearProgress(); 586 } 587 588 Optional<RpcCall> rpcCall = 589 matcher.isUserScan() ? RpcServer.getCurrentCall() : Optional.empty(); 590 591 int count = 0; 592 long totalBytesRead = 0; 593 // track the cells for metrics only if it is a user read request. 594 boolean onlyFromMemstore = matcher.isUserScan(); 595 try { 596 LOOP: do { 597 // Update and check the time limit based on the configured value of cellsPerTimeoutCheck 598 // Or if the preadMaxBytes is reached and we may want to return so we can switch to stream 599 // in 600 // the shipped method below. 601 if ( 602 kvsScanned % cellsPerHeartbeatCheck == 0 603 || (scanUsePread && readType == Scan.ReadType.DEFAULT && bytesRead > preadMaxBytes) 604 ) { 605 if (scannerContext.checkTimeLimit(LimitScope.BETWEEN_CELLS)) { 606 return scannerContext.setScannerState(NextState.TIME_LIMIT_REACHED).hasMoreValues(); 607 } 608 } 609 // Do object compare - we set prevKV from the same heap. 610 if (prevCell != cell) { 611 ++kvsScanned; 612 } 613 checkScanOrder(prevCell, cell, comparator); 614 int cellSize = PrivateCellUtil.estimatedSerializedSizeOf(cell); 615 bytesRead += cellSize; 616 if (scanUsePread && readType == Scan.ReadType.DEFAULT && bytesRead > preadMaxBytes) { 617 // return immediately if we want to switch from pread to stream. We need this because we 618 // can 619 // only switch in the shipped method, if user use a filter to filter out everything and 620 // rpc 621 // timeout is very large then the shipped method will never be called until the whole scan 622 // is finished, but at that time we have already scan all the data... 623 // See HBASE-20457 for more details. 624 // And there is still a scenario that can not be handled. If we have a very large row, 625 // which 626 // have millions of qualifiers, and filter.filterRow is used, then even if we set the flag 627 // here, we still need to scan all the qualifiers before returning... 628 scannerContext.returnImmediately(); 629 } 630 631 heap.recordBlockSize(blockSize -> { 632 if (rpcCall.isPresent()) { 633 rpcCall.get().incrementBlockBytesScanned(blockSize); 634 } 635 scannerContext.incrementBlockProgress(blockSize); 636 }); 637 638 prevCell = cell; 639 scannerContext.setLastPeekedCell(cell); 640 topChanged = false; 641 ScanQueryMatcher.MatchCode qcode = matcher.match(cell); 642 switch (qcode) { 643 case INCLUDE: 644 case INCLUDE_AND_SEEK_NEXT_ROW: 645 case INCLUDE_AND_SEEK_NEXT_COL: 646 Filter f = matcher.getFilter(); 647 if (f != null) { 648 Cell transformedCell = f.transformCell(cell); 649 // fast path, most filters just return the same cell instance 650 if (transformedCell != cell) { 651 if (transformedCell instanceof ExtendedCell) { 652 cell = (ExtendedCell) transformedCell; 653 } else { 654 throw new DoNotRetryIOException("Incorrect filter implementation, " 655 + "the Cell returned by transformCell is not an ExtendedCell. Filter class: " 656 + f.getClass().getName()); 657 } 658 } 659 } 660 this.countPerRow++; 661 662 // add to results only if we have skipped #storeOffset kvs 663 // also update metric accordingly 664 if (this.countPerRow > storeOffset) { 665 outResult.add(cell); 666 667 // Update local tracking information 668 count++; 669 totalBytesRead += cellSize; 670 671 /** 672 * Increment the metric if all the cells are from memstore. If not we will account it 673 * for mixed reads 674 */ 675 onlyFromMemstore = onlyFromMemstore && heap.isLatestCellFromMemstore(); 676 // Update the progress of the scanner context 677 scannerContext.incrementSizeProgress(cellSize, cell.heapSize()); 678 scannerContext.incrementBatchProgress(1); 679 680 if (matcher.isUserScan() && totalBytesRead > maxRowSize) { 681 String message = "Max row size allowed: " + maxRowSize 682 + ", but the row is bigger than that, the row info: " 683 + CellUtil.toString(cell, false) + ", already have process row cells = " 684 + outResult.size() + ", it belong to region = " 685 + store.getHRegion().getRegionInfo().getRegionNameAsString(); 686 LOG.warn(message); 687 throw new RowTooBigException(message); 688 } 689 690 if (storeLimit > -1 && this.countPerRow >= (storeLimit + storeOffset)) { 691 // do what SEEK_NEXT_ROW does. 692 if (!matcher.moreRowsMayExistAfter(cell)) { 693 close(false);// Do all cleanup except heap.close() 694 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 695 } 696 matcher.clearCurrentRow(); 697 seekToNextRow(cell); 698 break LOOP; 699 } 700 } 701 702 if (qcode == ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_ROW) { 703 if (!matcher.moreRowsMayExistAfter(cell)) { 704 close(false);// Do all cleanup except heap.close() 705 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 706 } 707 matcher.clearCurrentRow(); 708 seekOrSkipToNextRow(cell); 709 } else if (qcode == ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_COL) { 710 seekOrSkipToNextColumn(cell); 711 } else { 712 this.heap.next(); 713 } 714 715 if (scannerContext.checkBatchLimit(LimitScope.BETWEEN_CELLS)) { 716 break LOOP; 717 } 718 if (scannerContext.checkSizeLimit(LimitScope.BETWEEN_CELLS)) { 719 break LOOP; 720 } 721 continue; 722 723 case DONE: 724 // Optimization for Gets! If DONE, no more to get on this row, early exit! 725 if (get) { 726 // Then no more to this row... exit. 727 close(false);// Do all cleanup except heap.close() 728 // update metric 729 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 730 } 731 matcher.clearCurrentRow(); 732 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues(); 733 734 case DONE_SCAN: 735 close(false);// Do all cleanup except heap.close() 736 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 737 738 case SEEK_NEXT_ROW: 739 // This is just a relatively simple end of scan fix, to short-cut end 740 // us if there is an endKey in the scan. 741 if (!matcher.moreRowsMayExistAfter(cell)) { 742 close(false);// Do all cleanup except heap.close() 743 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 744 } 745 matcher.clearCurrentRow(); 746 seekOrSkipToNextRow(cell); 747 NextState stateAfterSeekNextRow = needToReturn(outResult); 748 if (stateAfterSeekNextRow != null) { 749 return scannerContext.setScannerState(stateAfterSeekNextRow).hasMoreValues(); 750 } 751 break; 752 753 case SEEK_NEXT_COL: 754 seekOrSkipToNextColumn(cell); 755 NextState stateAfterSeekNextColumn = needToReturn(outResult); 756 if (stateAfterSeekNextColumn != null) { 757 return scannerContext.setScannerState(stateAfterSeekNextColumn).hasMoreValues(); 758 } 759 break; 760 761 case SKIP: 762 this.heap.next(); 763 break; 764 765 case SEEK_NEXT_USING_HINT: 766 ExtendedCell nextKV = matcher.getNextKeyHint(cell); 767 if (nextKV != null) { 768 int difference = comparator.compare(nextKV, cell); 769 if ( 770 ((!scan.isReversed() && difference > 0) || (scan.isReversed() && difference < 0)) 771 ) { 772 seekAsDirection(nextKV); 773 NextState stateAfterSeekByHint = needToReturn(outResult); 774 if (stateAfterSeekByHint != null) { 775 return scannerContext.setScannerState(stateAfterSeekByHint).hasMoreValues(); 776 } 777 break; 778 } 779 } 780 heap.next(); 781 break; 782 783 default: 784 throw new RuntimeException("UNEXPECTED"); 785 } 786 787 // One last chance to break due to size limit. The INCLUDE* cases above already check 788 // limit and continue. For the various filtered cases, we need to check because block 789 // size limit may have been exceeded even if we don't add cells to result list. 790 if (scannerContext.checkSizeLimit(LimitScope.BETWEEN_CELLS)) { 791 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues(); 792 } 793 } while ((cell = this.heap.peek()) != null); 794 795 if (count > 0) { 796 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues(); 797 } 798 799 // No more keys 800 close(false);// Do all cleanup except heap.close() 801 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 802 } finally { 803 // increment only if we have some result 804 if (count > 0 && matcher.isUserScan()) { 805 // if true increment memstore metrics, if not the mixed one 806 updateMetricsStore(onlyFromMemstore); 807 } 808 } 809 } 810 811 private void updateMetricsStore(boolean memstoreRead) { 812 if (store != null) { 813 store.updateMetricsStore(memstoreRead); 814 } else { 815 // for testing. 816 if (memstoreRead) { 817 memstoreOnlyReads++; 818 } else { 819 mixedReads++; 820 } 821 } 822 } 823 824 /** 825 * If the top cell won't be flushed into disk, the new top cell may be changed after 826 * #reopenAfterFlush. Because the older top cell only exist in the memstore scanner but the 827 * memstore scanner is replaced by hfile scanner after #reopenAfterFlush. If the row of top cell 828 * is changed, we should return the current cells. Otherwise, we may return the cells across 829 * different rows. 830 * @param outResult the cells which are visible for user scan 831 * @return null is the top cell doesn't change. Otherwise, the NextState to return 832 */ 833 private NextState needToReturn(List<Cell> outResult) { 834 if (!outResult.isEmpty() && topChanged) { 835 return heap.peek() == null ? NextState.NO_MORE_VALUES : NextState.MORE_VALUES; 836 } 837 return null; 838 } 839 840 private void seekOrSkipToNextRow(ExtendedCell cell) throws IOException { 841 // If it is a Get Scan, then we know that we are done with this row; there are no more 842 // rows beyond the current one: don't try to optimize. 843 if (!get) { 844 if (trySkipToNextRow(cell)) { 845 return; 846 } 847 } 848 seekToNextRow(cell); 849 } 850 851 private void seekOrSkipToNextColumn(ExtendedCell cell) throws IOException { 852 if (!trySkipToNextColumn(cell)) { 853 seekAsDirection(matcher.getKeyForNextColumn(cell)); 854 } 855 } 856 857 /** 858 * See if we should actually SEEK or rather just SKIP to the next Cell (see HBASE-13109). 859 * ScanQueryMatcher may issue SEEK hints, such as seek to next column, next row, or seek to an 860 * arbitrary seek key. This method decides whether a seek is the most efficient _actual_ way to 861 * get us to the requested cell (SEEKs are more expensive than SKIP, SKIP, SKIP inside the 862 * current, loaded block). It does this by looking at the next indexed key of the current HFile. 863 * This key is then compared with the _SEEK_ key, where a SEEK key is an artificial 'last possible 864 * key on the row' (only in here, we avoid actually creating a SEEK key; in the compare we work 865 * with the current Cell but compare as though it were a seek key; see down in 866 * matcher.compareKeyForNextRow, etc). If the compare gets us onto the next block we *_SEEK, 867 * otherwise we just SKIP to the next requested cell. 868 * <p> 869 * Other notes: 870 * <ul> 871 * <li>Rows can straddle block boundaries</li> 872 * <li>Versions of columns can straddle block boundaries (i.e. column C1 at T1 might be in a 873 * different block than column C1 at T2)</li> 874 * <li>We want to SKIP if the chance is high that we'll find the desired Cell after a few 875 * SKIPs...</li> 876 * <li>We want to SEEK when the chance is high that we'll be able to seek past many Cells, 877 * especially if we know we need to go to the next block.</li> 878 * </ul> 879 * <p> 880 * A good proxy (best effort) to determine whether SKIP is better than SEEK is whether we'll 881 * likely end up seeking to the next block (or past the next block) to get our next column. 882 * Example: 883 * 884 * <pre> 885 * | BLOCK 1 | BLOCK 2 | 886 * | r1/c1, r1/c2, r1/c3 | r1/c4, r1/c5, r2/c1 | 887 * ^ ^ 888 * | | 889 * Next Index Key SEEK_NEXT_ROW (before r2/c1) 890 * 891 * 892 * | BLOCK 1 | BLOCK 2 | 893 * | r1/c1/t5, r1/c1/t4, r1/c1/t3 | r1/c1/t2, r1/c1/T1, r1/c2/T3 | 894 * ^ ^ 895 * | | 896 * Next Index Key SEEK_NEXT_COL 897 * </pre> 898 * 899 * Now imagine we want columns c1 and c3 (see first diagram above), the 'Next Index Key' of r1/c4 900 * is > r1/c3 so we should seek to get to the c1 on the next row, r2. In second case, say we only 901 * want one version of c1, after we have it, a SEEK_COL will be issued to get to c2. Looking at 902 * the 'Next Index Key', it would land us in the next block, so we should SEEK. In other scenarios 903 * where the SEEK will not land us in the next block, it is very likely better to issues a series 904 * of SKIPs. 905 * @param cell current cell 906 * @return true means skip to next row, false means not 907 */ 908 protected boolean trySkipToNextRow(ExtendedCell cell) throws IOException { 909 ExtendedCell nextCell = null; 910 // used to guard against a changed next indexed key by doing a identity comparison 911 // when the identity changes we need to compare the bytes again 912 ExtendedCell previousIndexedKey = null; 913 do { 914 ExtendedCell nextIndexedKey = getNextIndexedKey(); 915 if ( 916 nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY 917 && (nextIndexedKey == previousIndexedKey 918 || matcher.compareKeyForNextRow(nextIndexedKey, cell) >= 0) 919 ) { 920 this.heap.next(); 921 ++kvsScanned; 922 previousIndexedKey = nextIndexedKey; 923 } else { 924 return false; 925 } 926 } while ((nextCell = this.heap.peek()) != null && CellUtil.matchingRows(cell, nextCell)); 927 return true; 928 } 929 930 /** 931 * See {@link #trySkipToNextRow(ExtendedCell)} 932 * @param cell current cell 933 * @return true means skip to next column, false means not 934 */ 935 protected boolean trySkipToNextColumn(ExtendedCell cell) throws IOException { 936 ExtendedCell nextCell = null; 937 // used to guard against a changed next indexed key by doing a identity comparison 938 // when the identity changes we need to compare the bytes again 939 ExtendedCell previousIndexedKey = null; 940 do { 941 ExtendedCell nextIndexedKey = getNextIndexedKey(); 942 if ( 943 nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY 944 && (nextIndexedKey == previousIndexedKey 945 || matcher.compareKeyForNextColumn(nextIndexedKey, cell) >= 0) 946 ) { 947 this.heap.next(); 948 ++kvsScanned; 949 previousIndexedKey = nextIndexedKey; 950 } else { 951 return false; 952 } 953 } while ((nextCell = this.heap.peek()) != null && CellUtil.matchingRowColumn(cell, nextCell)); 954 // We need this check because it may happen that the new scanner that we get 955 // during heap.next() is requiring reseek due of fake KV previously generated for 956 // ROWCOL bloom filter optimization. See HBASE-19863 for more details 957 if ( 958 useRowColBloom && nextCell != null && cell.getTimestamp() == PrivateConstants.OLDEST_TIMESTAMP 959 ) { 960 return false; 961 } 962 return true; 963 } 964 965 @Override 966 public long getReadPoint() { 967 return this.readPt; 968 } 969 970 private static void clearAndClose(List<KeyValueScanner> scanners) { 971 if (scanners == null) { 972 return; 973 } 974 for (KeyValueScanner s : scanners) { 975 s.close(); 976 } 977 scanners.clear(); 978 } 979 980 // Implementation of ChangedReadersObserver 981 @Override 982 public void updateReaders(List<HStoreFile> sfs, List<KeyValueScanner> memStoreScanners) 983 throws IOException { 984 if (CollectionUtils.isEmpty(sfs) && CollectionUtils.isEmpty(memStoreScanners)) { 985 return; 986 } 987 boolean updateReaders = false; 988 flushLock.lock(); 989 try { 990 if (!closeLock.tryLock()) { 991 // The reason for doing this is that when the current store scanner does not retrieve 992 // any new cells, then the scanner is considered to be done. The heap of this scanner 993 // is not closed till the shipped() call is completed. Hence in that case if at all 994 // the partial close (close (false)) has been called before updateReaders(), there is no 995 // need for the updateReaders() to happen. 996 LOG.debug("StoreScanner already has the close lock. There is no need to updateReaders"); 997 // no lock acquired. 998 clearAndClose(memStoreScanners); 999 return; 1000 } 1001 // lock acquired 1002 updateReaders = true; 1003 if (this.closing) { 1004 LOG.debug("StoreScanner already closing. There is no need to updateReaders"); 1005 clearAndClose(memStoreScanners); 1006 return; 1007 } 1008 flushed = true; 1009 final boolean isCompaction = false; 1010 boolean usePread = get || scanUsePread; 1011 // SEE HBASE-19468 where the flushed files are getting compacted even before a scanner 1012 // calls next(). So its better we create scanners here rather than next() call. Ensure 1013 // these scanners are properly closed() whether or not the scan is completed successfully 1014 // Eagerly creating scanners so that we have the ref counting ticking on the newly created 1015 // store files. In case of stream scanners this eager creation does not induce performance 1016 // penalty because in scans (that uses stream scanners) the next() call is bound to happen. 1017 List<KeyValueScanner> scanners = 1018 store.getScanners(sfs, cacheBlocks, get, usePread, isCompaction, matcher, 1019 scan.getStartRow(), scan.getStopRow(), this.readPt, false, isOnlyLatestVersionScan(scan)); 1020 flushedstoreFileScanners.addAll(scanners); 1021 if (!CollectionUtils.isEmpty(memStoreScanners)) { 1022 clearAndClose(memStoreScannersAfterFlush); 1023 memStoreScannersAfterFlush.addAll(memStoreScanners); 1024 } 1025 } finally { 1026 flushLock.unlock(); 1027 if (updateReaders) { 1028 closeLock.unlock(); 1029 } 1030 } 1031 // Let the next() call handle re-creating and seeking 1032 } 1033 1034 /** Returns if top of heap has changed (and KeyValueHeap has to try the next KV) */ 1035 protected final boolean reopenAfterFlush() throws IOException { 1036 // here we can make sure that we have a Store instance so no null check on store. 1037 ExtendedCell lastTop = heap.peek(); 1038 // When we have the scan object, should we not pass it to getScanners() to get a limited set of 1039 // scanners? We did so in the constructor and we could have done it now by storing the scan 1040 // object from the constructor 1041 List<KeyValueScanner> scanners; 1042 flushLock.lock(); 1043 try { 1044 List<KeyValueScanner> allScanners = 1045 new ArrayList<>(flushedstoreFileScanners.size() + memStoreScannersAfterFlush.size()); 1046 allScanners.addAll(flushedstoreFileScanners); 1047 allScanners.addAll(memStoreScannersAfterFlush); 1048 scanners = selectScannersFrom(store, allScanners); 1049 // Clear the current set of flushed store files scanners so that they don't get added again 1050 flushedstoreFileScanners.clear(); 1051 memStoreScannersAfterFlush.clear(); 1052 } finally { 1053 flushLock.unlock(); 1054 } 1055 1056 // Seek the new scanners to the last key 1057 seekScanners(scanners, lastTop, false, parallelSeekEnabled); 1058 // remove the older memstore scanner 1059 for (int i = currentScanners.size() - 1; i >= 0; i--) { 1060 if (!currentScanners.get(i).isFileScanner()) { 1061 scannersForDelayedClose.add(currentScanners.remove(i)); 1062 } else { 1063 // we add the memstore scanner to the end of currentScanners 1064 break; 1065 } 1066 } 1067 // add the newly created scanners on the flushed files and the current active memstore scanner 1068 addCurrentScanners(scanners); 1069 // Combine all seeked scanners with a heap 1070 resetKVHeap(this.currentScanners, store.getComparator()); 1071 resetQueryMatcher(lastTop); 1072 if (heap.peek() == null || store.getComparator().compareRows(lastTop, this.heap.peek()) != 0) { 1073 LOG.info("Storescanner.peek() is changed where before = " + lastTop.toString() 1074 + ",and after = " + heap.peek()); 1075 topChanged = true; 1076 } else { 1077 topChanged = false; 1078 } 1079 return topChanged; 1080 } 1081 1082 private void resetQueryMatcher(ExtendedCell lastTopKey) { 1083 // Reset the state of the Query Matcher and set to top row. 1084 // Only reset and call setRow if the row changes; avoids confusing the 1085 // query matcher if scanning intra-row. 1086 ExtendedCell cell = heap.peek(); 1087 if (cell == null) { 1088 cell = lastTopKey; 1089 } 1090 if ((matcher.currentRow() == null) || !CellUtil.matchingRows(cell, matcher.currentRow())) { 1091 this.countPerRow = 0; 1092 // The setToNewRow will call reset internally 1093 matcher.setToNewRow(cell); 1094 } 1095 } 1096 1097 /** 1098 * Check whether scan as expected order 1099 */ 1100 protected void checkScanOrder(Cell prevKV, Cell kv, CellComparator comparator) 1101 throws IOException { 1102 // Check that the heap gives us KVs in an increasing order. 1103 assert prevKV == null || comparator == null || comparator.compare(prevKV, kv) <= 0 1104 : "Key " + prevKV + " followed by a smaller key " + kv + " in cf " + store; 1105 } 1106 1107 protected boolean seekToNextRow(ExtendedCell c) throws IOException { 1108 return reseek(PrivateCellUtil.createLastOnRow(c)); 1109 } 1110 1111 /** 1112 * Do a reseek in a normal StoreScanner(scan forward) 1113 * @return true if scanner has values left, false if end of scanner 1114 */ 1115 protected boolean seekAsDirection(ExtendedCell kv) throws IOException { 1116 return reseek(kv); 1117 } 1118 1119 @Override 1120 public boolean reseek(ExtendedCell kv) throws IOException { 1121 if (checkFlushed()) { 1122 reopenAfterFlush(); 1123 } 1124 if (explicitColumnQuery && lazySeekEnabledGlobally) { 1125 return heap.requestSeek(kv, true, useRowColBloom); 1126 } 1127 return heap.reseek(kv); 1128 } 1129 1130 void trySwitchToStreamRead() { 1131 if ( 1132 readType != Scan.ReadType.DEFAULT || !scanUsePread || closing || heap.peek() == null 1133 || bytesRead < preadMaxBytes 1134 ) { 1135 return; 1136 } 1137 LOG.debug("Switch to stream read (scanned={} bytes) of {}", bytesRead, 1138 this.store.getColumnFamilyName()); 1139 scanUsePread = false; 1140 ExtendedCell lastTop = heap.peek(); 1141 List<KeyValueScanner> memstoreScanners = new ArrayList<>(); 1142 List<KeyValueScanner> scannersToClose = new ArrayList<>(); 1143 for (KeyValueScanner kvs : currentScanners) { 1144 if (!kvs.isFileScanner()) { 1145 // collect memstorescanners here 1146 memstoreScanners.add(kvs); 1147 } else { 1148 scannersToClose.add(kvs); 1149 } 1150 } 1151 List<KeyValueScanner> fileScanners = null; 1152 List<KeyValueScanner> newCurrentScanners; 1153 KeyValueHeap newHeap; 1154 try { 1155 // We must have a store instance here so no null check 1156 // recreate the scanners on the current file scanners 1157 fileScanners = store.recreateScanners(scannersToClose, cacheBlocks, false, false, matcher, 1158 scan.getStartRow(), scan.includeStartRow(), scan.getStopRow(), scan.includeStopRow(), 1159 readPt, false); 1160 if (fileScanners == null) { 1161 return; 1162 } 1163 seekScanners(fileScanners, lastTop, false, parallelSeekEnabled); 1164 newCurrentScanners = new ArrayList<>(fileScanners.size() + memstoreScanners.size()); 1165 newCurrentScanners.addAll(fileScanners); 1166 newCurrentScanners.addAll(memstoreScanners); 1167 newHeap = newKVHeap(newCurrentScanners, comparator); 1168 } catch (Exception e) { 1169 LOG.warn("failed to switch to stream read", e); 1170 if (fileScanners != null) { 1171 fileScanners.forEach(KeyValueScanner::close); 1172 } 1173 return; 1174 } 1175 currentScanners.clear(); 1176 addCurrentScanners(newCurrentScanners); 1177 this.heap = newHeap; 1178 resetQueryMatcher(lastTop); 1179 scannersToClose.forEach(KeyValueScanner::close); 1180 } 1181 1182 protected final boolean checkFlushed() { 1183 // check the var without any lock. Suppose even if we see the old 1184 // value here still it is ok to continue because we will not be resetting 1185 // the heap but will continue with the referenced memstore's snapshot. For compactions 1186 // any way we don't need the updateReaders at all to happen as we still continue with 1187 // the older files 1188 if (flushed) { 1189 // If there is a flush and the current scan is notified on the flush ensure that the 1190 // scan's heap gets reset and we do a seek on the newly flushed file. 1191 if (this.closing) { 1192 return false; 1193 } 1194 // reset the flag 1195 flushed = false; 1196 return true; 1197 } 1198 return false; 1199 } 1200 1201 /** 1202 * Seek storefiles in parallel to optimize IO latency as much as possible 1203 * @param scanners the list {@link KeyValueScanner}s to be read from 1204 * @param kv the KeyValue on which the operation is being requested 1205 */ 1206 private void parallelSeek(final List<? extends KeyValueScanner> scanners, final ExtendedCell kv) 1207 throws IOException { 1208 if (scanners.isEmpty()) return; 1209 int storeFileScannerCount = scanners.size(); 1210 CountDownLatch latch = new CountDownLatch(storeFileScannerCount); 1211 List<ParallelSeekHandler> handlers = new ArrayList<>(storeFileScannerCount); 1212 for (KeyValueScanner scanner : scanners) { 1213 if (scanner instanceof StoreFileScanner) { 1214 ParallelSeekHandler seekHandler = new ParallelSeekHandler(scanner, kv, this.readPt, latch); 1215 executor.submit(seekHandler); 1216 handlers.add(seekHandler); 1217 } else { 1218 scanner.seek(kv); 1219 latch.countDown(); 1220 } 1221 } 1222 1223 try { 1224 latch.await(); 1225 } catch (InterruptedException ie) { 1226 throw (InterruptedIOException) new InterruptedIOException().initCause(ie); 1227 } 1228 1229 for (ParallelSeekHandler handler : handlers) { 1230 if (handler.getErr() != null) { 1231 throw new IOException(handler.getErr()); 1232 } 1233 } 1234 } 1235 1236 /** 1237 * Used in testing. 1238 * @return all scanners in no particular order 1239 */ 1240 List<KeyValueScanner> getAllScannersForTesting() { 1241 List<KeyValueScanner> allScanners = new ArrayList<>(); 1242 KeyValueScanner current = heap.getCurrentForTesting(); 1243 if (current != null) allScanners.add(current); 1244 for (KeyValueScanner scanner : heap.getHeap()) 1245 allScanners.add(scanner); 1246 return allScanners; 1247 } 1248 1249 static void enableLazySeekGlobally(boolean enable) { 1250 lazySeekEnabledGlobally = enable; 1251 } 1252 1253 /** Returns The estimated number of KVs seen by this scanner (includes some skipped KVs). */ 1254 public long getEstimatedNumberOfKvsScanned() { 1255 return this.kvsScanned; 1256 } 1257 1258 @Override 1259 public ExtendedCell getNextIndexedKey() { 1260 return this.heap.getNextIndexedKey(); 1261 } 1262 1263 @Override 1264 public void shipped() throws IOException { 1265 if (prevCell != null) { 1266 // Do the copy here so that in case the prevCell ref is pointing to the previous 1267 // blocks we can safely release those blocks. 1268 // This applies to blocks that are got from Bucket cache, L1 cache and the blocks 1269 // fetched from HDFS. Copying this would ensure that we let go the references to these 1270 // blocks so that they can be GCed safely(in case of bucket cache) 1271 prevCell = KeyValueUtil.toNewKeyCell(this.prevCell); 1272 } 1273 matcher.beforeShipped(); 1274 // There wont be further fetch of Cells from these scanners. Just close. 1275 clearAndClose(scannersForDelayedClose); 1276 if (this.heap != null) { 1277 this.heap.shipped(); 1278 // When switching from pread to stream, we will open a new scanner for each store file, but 1279 // the old scanner may still track the HFileBlocks we have scanned but not sent back to client 1280 // yet. If we close the scanner immediately then the HFileBlocks may be messed up by others 1281 // before we serialize and send it back to client. The HFileBlocks will be released in shipped 1282 // method, so we here will also open new scanners and close old scanners in shipped method. 1283 // See HBASE-18055 for more details. 1284 trySwitchToStreamRead(); 1285 } 1286 } 1287}