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.util.AbstractList; 022import java.util.ArrayList; 023import java.util.Collection; 024import java.util.Collections; 025import java.util.HashSet; 026import java.util.List; 027import java.util.Map; 028import java.util.NavigableSet; 029import java.util.Optional; 030import java.util.Set; 031import java.util.concurrent.ConcurrentHashMap; 032import org.apache.hadoop.fs.Path; 033import org.apache.hadoop.hbase.Cell; 034import org.apache.hadoop.hbase.CellComparator; 035import org.apache.hadoop.hbase.CellUtil; 036import org.apache.hadoop.hbase.DoNotRetryIOException; 037import org.apache.hadoop.hbase.ExtendedCell; 038import org.apache.hadoop.hbase.HConstants; 039import org.apache.hadoop.hbase.KeyValue; 040import org.apache.hadoop.hbase.PrivateCellUtil; 041import org.apache.hadoop.hbase.UnknownScannerException; 042import org.apache.hadoop.hbase.client.ClientInternalHelper; 043import org.apache.hadoop.hbase.client.IsolationLevel; 044import org.apache.hadoop.hbase.client.RegionInfo; 045import org.apache.hadoop.hbase.client.Scan; 046import org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics; 047import org.apache.hadoop.hbase.filter.FilterWrapper; 048import org.apache.hadoop.hbase.filter.IncompatibleFilterException; 049import org.apache.hadoop.hbase.ipc.CallerDisconnectedException; 050import org.apache.hadoop.hbase.ipc.RpcCall; 051import org.apache.hadoop.hbase.ipc.RpcCallback; 052import org.apache.hadoop.hbase.ipc.RpcServer; 053import org.apache.hadoop.hbase.regionserver.Region.Operation; 054import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope; 055import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState; 056import org.apache.hadoop.hbase.trace.TraceUtil; 057import org.apache.hadoop.hbase.util.Bytes; 058import org.apache.yetus.audience.InterfaceAudience; 059import org.slf4j.Logger; 060import org.slf4j.LoggerFactory; 061 062import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; 063 064/** 065 * RegionScannerImpl is used to combine scanners from multiple Stores (aka column families). 066 */ 067@InterfaceAudience.Private 068public class RegionScannerImpl implements RegionScanner, Shipper, RpcCallback { 069 070 private static final Logger LOG = LoggerFactory.getLogger(RegionScannerImpl.class); 071 072 // Package local for testability 073 KeyValueHeap storeHeap = null; 074 075 /** 076 * Heap of key-values that are not essential for the provided filters and are thus read on demand, 077 * if on-demand column family loading is enabled. 078 */ 079 KeyValueHeap joinedHeap = null; 080 081 /** 082 * If the joined heap data gathering is interrupted due to scan limits, this will contain the row 083 * for which we are populating the values. 084 */ 085 protected ExtendedCell joinedContinuationRow = null; 086 private boolean filterClosed = false; 087 088 protected final byte[] stopRow; 089 protected final boolean includeStopRow; 090 protected final boolean reversed; 091 protected final HRegion region; 092 protected final CellComparator comparator; 093 094 private final ConcurrentHashMap<RegionScanner, Long> scannerReadPoints; 095 096 private final long readPt; 097 private final long maxResultSize; 098 private final ScannerContext defaultScannerContext; 099 private final FilterWrapper filter; 100 private final String operationId; 101 102 private RegionServerServices rsServices; 103 104 private final Set<Path> filesRead = new HashSet<>(); 105 106 @Override 107 public RegionInfo getRegionInfo() { 108 return region.getRegionInfo(); 109 } 110 111 private static boolean hasNonce(HRegion region, long nonce) { 112 RegionServerServices rsServices = region.getRegionServerServices(); 113 return nonce != HConstants.NO_NONCE && rsServices != null 114 && rsServices.getNonceManager() != null; 115 } 116 117 RegionScannerImpl(Scan scan, List<KeyValueScanner> additionalScanners, HRegion region, 118 long nonceGroup, long nonce) throws IOException { 119 this.region = region; 120 this.maxResultSize = scan.getMaxResultSize(); 121 if (scan.hasFilter()) { 122 this.filter = new FilterWrapper(scan.getFilter()); 123 } else { 124 this.filter = null; 125 } 126 this.comparator = region.getCellComparator(); 127 /** 128 * By default, calls to next/nextRaw must enforce the batch limit. Thus, construct a default 129 * scanner context that can be used to enforce the batch limit in the event that a 130 * ScannerContext is not specified during an invocation of next/nextRaw 131 */ 132 defaultScannerContext = ScannerContext.newBuilder().setBatchLimit(scan.getBatch()).build(); 133 this.stopRow = scan.getStopRow(); 134 this.includeStopRow = scan.includeStopRow(); 135 this.reversed = scan.isReversed(); 136 this.operationId = scan.getId(); 137 138 // synchronize on scannerReadPoints so that nobody calculates 139 // getSmallestReadPoint, before scannerReadPoints is updated. 140 IsolationLevel isolationLevel = scan.getIsolationLevel(); 141 long mvccReadPoint = ClientInternalHelper.getMvccReadPoint(scan); 142 this.scannerReadPoints = region.scannerReadPoints; 143 this.rsServices = region.getRegionServerServices(); 144 region.smallestReadPointCalcLock.lock(ReadPointCalculationLock.LockType.RECORDING_LOCK); 145 try { 146 if (mvccReadPoint > 0) { 147 this.readPt = mvccReadPoint; 148 } else if (hasNonce(region, nonce)) { 149 this.readPt = rsServices.getNonceManager().getMvccFromOperationContext(nonceGroup, nonce); 150 } else { 151 this.readPt = region.getReadPoint(isolationLevel); 152 } 153 scannerReadPoints.put(this, this.readPt); 154 } finally { 155 region.smallestReadPointCalcLock.unlock(ReadPointCalculationLock.LockType.RECORDING_LOCK); 156 } 157 initializeScanners(scan, additionalScanners); 158 } 159 160 public ScannerContext getContext() { 161 return defaultScannerContext; 162 } 163 164 private void initializeScanners(Scan scan, List<KeyValueScanner> additionalScanners) 165 throws IOException { 166 // Here we separate all scanners into two lists - scanner that provide data required 167 // by the filter to operate (scanners list) and all others (joinedScanners list). 168 List<KeyValueScanner> scanners = new ArrayList<>(scan.getFamilyMap().size()); 169 List<KeyValueScanner> joinedScanners = new ArrayList<>(scan.getFamilyMap().size()); 170 // Store all already instantiated scanners for exception handling 171 List<KeyValueScanner> instantiatedScanners = new ArrayList<>(); 172 // handle additionalScanners 173 if (additionalScanners != null && !additionalScanners.isEmpty()) { 174 scanners.addAll(additionalScanners); 175 instantiatedScanners.addAll(additionalScanners); 176 } 177 178 try { 179 for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap().entrySet()) { 180 HStore store = region.getStore(entry.getKey()); 181 KeyValueScanner scanner = store.getScanner(scan, entry.getValue(), this.readPt); 182 instantiatedScanners.add(scanner); 183 if ( 184 this.filter == null || !scan.doLoadColumnFamiliesOnDemand() 185 || this.filter.isFamilyEssential(entry.getKey()) 186 ) { 187 scanners.add(scanner); 188 } else { 189 joinedScanners.add(scanner); 190 } 191 } 192 initializeKVHeap(scanners, joinedScanners, region); 193 } catch (Throwable t) { 194 throw handleException(instantiatedScanners, t); 195 } 196 } 197 198 protected void initializeKVHeap(List<KeyValueScanner> scanners, 199 List<KeyValueScanner> joinedScanners, HRegion region) throws IOException { 200 this.storeHeap = new KeyValueHeap(scanners, comparator); 201 if (!joinedScanners.isEmpty()) { 202 this.joinedHeap = new KeyValueHeap(joinedScanners, comparator); 203 } 204 } 205 206 private IOException handleException(List<KeyValueScanner> instantiatedScanners, Throwable t) { 207 // remove scaner read point before throw the exception 208 scannerReadPoints.remove(this); 209 if (storeHeap != null) { 210 storeHeap.close(); 211 storeHeap = null; 212 if (joinedHeap != null) { 213 joinedHeap.close(); 214 joinedHeap = null; 215 } 216 } else { 217 // close all already instantiated scanners before throwing the exception 218 for (KeyValueScanner scanner : instantiatedScanners) { 219 scanner.close(); 220 } 221 } 222 return t instanceof IOException ? (IOException) t : new IOException(t); 223 } 224 225 @Override 226 public long getMaxResultSize() { 227 return maxResultSize; 228 } 229 230 @Override 231 public long getMvccReadPoint() { 232 return this.readPt; 233 } 234 235 @Override 236 public int getBatch() { 237 return this.defaultScannerContext.getBatchLimit(); 238 } 239 240 @Override 241 public String getOperationId() { 242 return operationId; 243 } 244 245 /** 246 * Reset both the filter and the old filter. 247 * @throws IOException in case a filter raises an I/O exception. 248 */ 249 protected final void resetFilters() throws IOException { 250 if (filter != null) { 251 filter.reset(); 252 } 253 } 254 255 @Override 256 public boolean next(List<? super ExtendedCell> outResults) throws IOException { 257 // apply the batching limit by default 258 return next(outResults, defaultScannerContext); 259 } 260 261 @Override 262 public synchronized boolean next(List<? super ExtendedCell> outResults, 263 ScannerContext scannerContext) throws IOException { 264 if (this.filterClosed) { 265 throw new UnknownScannerException("Scanner was closed (timed out?) " 266 + "after we renewed it. Could be caused by a very slow scanner " 267 + "or a lengthy garbage collection"); 268 } 269 region.startRegionOperation(Operation.SCAN); 270 try { 271 return nextRaw(outResults, scannerContext); 272 } finally { 273 region.closeRegionOperation(Operation.SCAN); 274 } 275 } 276 277 @Override 278 public boolean nextRaw(List<? super ExtendedCell> outResults) throws IOException { 279 // Use the RegionScanner's context by default 280 return nextRaw(outResults, defaultScannerContext); 281 } 282 283 @Override 284 public boolean nextRaw(List<? super ExtendedCell> outResults, ScannerContext scannerContext) 285 throws IOException { 286 if (storeHeap == null) { 287 // scanner is closed 288 throw new UnknownScannerException("Scanner was closed"); 289 } 290 boolean moreValues = false; 291 if (outResults.isEmpty()) { 292 // Usually outResults is empty. This is true when next is called 293 // to handle scan or get operation. 294 moreValues = nextInternal(outResults, scannerContext); 295 } else { 296 List<ExtendedCell> tmpList = new ArrayList<>(); 297 moreValues = nextInternal(tmpList, scannerContext); 298 outResults.addAll(tmpList); 299 } 300 region.addReadRequestsCount(1); 301 if (region.getMetrics() != null) { 302 region.getMetrics().updateReadRequestCount(); 303 } 304 305 // If the size limit was reached it means a partial Result is being returned. Returning a 306 // partial Result means that we should not reset the filters; filters should only be reset in 307 // between rows 308 if (!scannerContext.mayHaveMoreCellsInRow()) { 309 resetFilters(); 310 } 311 312 if (isFilterDoneInternal()) { 313 moreValues = false; 314 } 315 return moreValues; 316 } 317 318 /** Returns true if more cells exist after this batch, false if scanner is done */ 319 private boolean populateFromJoinedHeap(List<? super ExtendedCell> results, 320 ScannerContext scannerContext) throws IOException { 321 assert joinedContinuationRow != null; 322 boolean moreValues = 323 populateResult(results, this.joinedHeap, scannerContext, joinedContinuationRow); 324 325 if (!scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) { 326 // We are done with this row, reset the continuation. 327 joinedContinuationRow = null; 328 } 329 // As the data is obtained from two independent heaps, we need to 330 // ensure that result list is sorted, because Result relies on that. 331 ((List<Cell>) results).sort(comparator); 332 return moreValues; 333 } 334 335 /** 336 * Fetches records with currentRow into results list, until next row, batchLimit (if not -1) is 337 * reached, or remainingResultSize (if not -1) is reaced 338 * @param heap KeyValueHeap to fetch data from.It must be positioned on correct row before call. 339 * @return state of last call to {@link KeyValueHeap#next()} 340 */ 341 private boolean populateResult(List<? super ExtendedCell> results, KeyValueHeap heap, 342 ScannerContext scannerContext, ExtendedCell currentRowCell) throws IOException { 343 Cell nextKv; 344 boolean moreCellsInRow = false; 345 boolean tmpKeepProgress = scannerContext.getKeepProgress(); 346 // Scanning between column families and thus the scope is between cells 347 LimitScope limitScope = LimitScope.BETWEEN_CELLS; 348 do { 349 // Check for thread interrupt status in case we have been signaled from 350 // #interruptRegionOperation. 351 region.checkInterrupt(); 352 353 // We want to maintain any progress that is made towards the limits while scanning across 354 // different column families. To do this, we toggle the keep progress flag on during calls 355 // to the StoreScanner to ensure that any progress made thus far is not wiped away. 356 scannerContext.setKeepProgress(true); 357 heap.next(results, scannerContext); 358 scannerContext.setKeepProgress(tmpKeepProgress); 359 360 nextKv = heap.peek(); 361 moreCellsInRow = moreCellsInRow(nextKv, currentRowCell); 362 // A row is scanned once its cells have been read from the store heap. The joined heap only 363 // re-populates rows that already passed the filter with the cells of the non essential 364 // families (HBASE-5416), so counting a completed row there would double count it. 365 if (!moreCellsInRow && heap == this.storeHeap) { 366 incrementCountOfRowsScannedMetric(scannerContext); 367 } 368 if (moreCellsInRow && scannerContext.checkBatchLimit(limitScope)) { 369 return scannerContext.setScannerState(NextState.BATCH_LIMIT_REACHED).hasMoreValues(); 370 } else if (scannerContext.checkSizeLimit(limitScope)) { 371 ScannerContext.NextState state = 372 moreCellsInRow ? NextState.SIZE_LIMIT_REACHED_MID_ROW : NextState.SIZE_LIMIT_REACHED; 373 return scannerContext.setScannerState(state).hasMoreValues(); 374 } else if (scannerContext.checkTimeLimit(limitScope)) { 375 ScannerContext.NextState state = 376 moreCellsInRow ? NextState.TIME_LIMIT_REACHED_MID_ROW : NextState.TIME_LIMIT_REACHED; 377 return scannerContext.setScannerState(state).hasMoreValues(); 378 } 379 } while (moreCellsInRow); 380 return nextKv != null; 381 } 382 383 /** 384 * Based on the nextKv in the heap, and the current row, decide whether or not there are more 385 * cells to be read in the heap. If the row of the nextKv in the heap matches the current row then 386 * there are more cells to be read in the row. 387 * @return true When there are more cells in the row to be read 388 */ 389 private boolean moreCellsInRow(final Cell nextKv, Cell currentRowCell) { 390 return nextKv != null && CellUtil.matchingRows(nextKv, currentRowCell); 391 } 392 393 /** Returns True if a filter rules the scanner is over, done. */ 394 @Override 395 public synchronized boolean isFilterDone() throws IOException { 396 return isFilterDoneInternal(); 397 } 398 399 private boolean isFilterDoneInternal() throws IOException { 400 return this.filter != null && this.filter.filterAllRemaining(); 401 } 402 403 private void checkClientDisconnect(Optional<RpcCall> rpcCall) throws CallerDisconnectedException { 404 if (rpcCall.isPresent()) { 405 // If a user specifies a too-restrictive or too-slow scanner, the 406 // client might time out and disconnect while the server side 407 // is still processing the request. We should abort aggressively 408 // in that case. 409 long afterTime = rpcCall.get().disconnectSince(); 410 if (afterTime >= 0) { 411 throw new CallerDisconnectedException( 412 "Aborting on region " + getRegionInfo().getRegionNameAsString() + ", call " + this 413 + " after " + afterTime + " ms, since " + "caller disconnected"); 414 } 415 } 416 } 417 418 private void resetProgress(ScannerContext scannerContext, int initialBatchProgress, 419 long initialSizeProgress, long initialHeapSizeProgress) { 420 // Starting to scan a new row. Reset the scanner progress according to whether or not 421 // progress should be kept. 422 if (scannerContext.getKeepProgress()) { 423 // Progress should be kept. Reset to initial values seen at start of method invocation. 424 scannerContext.setProgress(initialBatchProgress, initialSizeProgress, 425 initialHeapSizeProgress); 426 } else { 427 scannerContext.clearProgress(); 428 } 429 } 430 431 private boolean nextInternal(List<? super ExtendedCell> results, ScannerContext scannerContext) 432 throws IOException { 433 Preconditions.checkArgument(results.isEmpty(), "First parameter should be an empty list"); 434 Preconditions.checkArgument(scannerContext != null, "Scanner context cannot be null"); 435 Optional<RpcCall> rpcCall = RpcServer.getCurrentCall(); 436 437 // Save the initial progress from the Scanner context in these local variables. The progress 438 // may need to be reset a few times if rows are being filtered out so we save the initial 439 // progress. 440 int initialBatchProgress = scannerContext.getBatchProgress(); 441 long initialSizeProgress = scannerContext.getDataSizeProgress(); 442 long initialHeapSizeProgress = scannerContext.getHeapSizeProgress(); 443 444 // Used to check time limit 445 LimitScope limitScope = LimitScope.BETWEEN_CELLS; 446 447 // The loop here is used only when at some point during the next we determine 448 // that due to effects of filters or otherwise, we have an empty row in the result. 449 // Then we loop and try again. Otherwise, we must get out on the first iteration via return, 450 // "true" if there's more data to read, "false" if there isn't (storeHeap is at a stop row, 451 // and joinedHeap has no more data to read for the last row (if set, joinedContinuationRow). 452 while (true) { 453 resetProgress(scannerContext, initialBatchProgress, initialSizeProgress, 454 initialHeapSizeProgress); 455 checkClientDisconnect(rpcCall); 456 457 // Check for thread interrupt status in case we have been signaled from 458 // #interruptRegionOperation. 459 region.checkInterrupt(); 460 461 // Let's see what we have in the storeHeap. 462 ExtendedCell current = this.storeHeap.peek(); 463 464 boolean shouldStop = shouldStop(current); 465 // When has filter row is true it means that the all the cells for a particular row must be 466 // read before a filtering decision can be made. This means that filters where hasFilterRow 467 // run the risk of enLongAddering out of memory errors in the case that they are applied to a 468 // table that has very large rows. 469 boolean hasFilterRow = this.filter != null && this.filter.hasFilterRow(); 470 471 // If filter#hasFilterRow is true, partial results are not allowed since allowing them 472 // would prevent the filters from being evaluated. Thus, if it is true, change the 473 // scope of any limits that could potentially create partial results to 474 // LimitScope.BETWEEN_ROWS so that those limits are not reached mid-row 475 if (hasFilterRow) { 476 if (LOG.isTraceEnabled()) { 477 LOG.trace("filter#hasFilterRow is true which prevents partial results from being " 478 + " formed. Changing scope of limits that may create partials"); 479 } 480 scannerContext.setSizeLimitScope(LimitScope.BETWEEN_ROWS); 481 scannerContext.setTimeLimitScope(LimitScope.BETWEEN_ROWS); 482 limitScope = LimitScope.BETWEEN_ROWS; 483 } 484 485 if (scannerContext.checkTimeLimit(LimitScope.BETWEEN_CELLS)) { 486 if (hasFilterRow) { 487 throw new IncompatibleFilterException( 488 "Filter whose hasFilterRow() returns true is incompatible with scans that must " 489 + " stop mid-row because of a limit. ScannerContext:" + scannerContext); 490 } 491 return true; 492 } 493 494 // Check if we were getting data from the joinedHeap and hit the limit. 495 // If not, then it's main path - getting results from storeHeap. 496 if (joinedContinuationRow == null) { 497 // First, check if we are at a stop row. If so, there are no more results. 498 if (shouldStop) { 499 if (hasFilterRow) { 500 filter.filterRowCells((List<Cell>) results); 501 } 502 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 503 } 504 505 // Check if rowkey filter wants to exclude this row. If so, loop to next. 506 // Technically, if we hit limits before on this row, we don't need this call. 507 if (filterRowKey(current)) { 508 incrementCountOfRowsFilteredMetric(scannerContext); 509 // early check, see HBASE-16296 510 if (isFilterDoneInternal()) { 511 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 512 } 513 // HBASE-29974: ask the filter for a seek hint so we can jump directly past the rejected 514 // row instead of iterating through its cells one-by-one via nextRow(). 515 ExtendedCell rowHint = getHintForRejectedRow(current); 516 // Typically the count of rows scanned is incremented inside #populateResult. However, 517 // here we are filtering a row based purely on its row key, preventing us from calling 518 // #populateResult. Thus, perform the necessary increment here to rows scanned metric. 519 // Placed after getHintForRejectedRow so that a buggy filter throwing DNRIOE doesn't 520 // leave the metric incremented for a row that was never actually processed. 521 incrementCountOfRowsScannedMetric(scannerContext); 522 boolean moreRows = (rowHint != null) 523 ? nextRowViaHint(scannerContext, current, rowHint) 524 : nextRow(scannerContext, current); 525 if (!moreRows) { 526 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 527 } 528 results.clear(); 529 530 // Read nothing as the rowkey was filtered, but still need to check time limit 531 // We also check size limit because we might have read blocks in getting to this point. 532 if (scannerContext.checkAnyLimitReached(limitScope)) { 533 return true; 534 } 535 continue; 536 } 537 538 // Ok, we are good, let's try to get some results from the main heap. 539 populateResult(results, this.storeHeap, scannerContext, current); 540 if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) { 541 if (hasFilterRow) { 542 throw new IncompatibleFilterException( 543 "Filter whose hasFilterRow() returns true is incompatible with scans that must " 544 + " stop mid-row because of a limit. ScannerContext:" + scannerContext); 545 } 546 return true; 547 } 548 549 // Check for thread interrupt status in case we have been signaled from 550 // #interruptRegionOperation. 551 region.checkInterrupt(); 552 553 Cell nextKv = this.storeHeap.peek(); 554 shouldStop = shouldStop(nextKv); 555 // save that the row was empty before filters applied to it. 556 final boolean isEmptyRow = results.isEmpty(); 557 558 // We have the part of the row necessary for filtering (all of it, usually). 559 // First filter with the filterRow(List). 560 FilterWrapper.FilterRowRetCode ret = FilterWrapper.FilterRowRetCode.NOT_CALLED; 561 if (hasFilterRow) { 562 ret = filter.filterRowCellsWithRet((List<Cell>) results); 563 564 // We don't know how the results have changed after being filtered. Must set progress 565 // according to contents of results now. 566 if (scannerContext.getKeepProgress()) { 567 scannerContext.setProgress(initialBatchProgress, initialSizeProgress, 568 initialHeapSizeProgress); 569 } else { 570 scannerContext.clearProgress(); 571 } 572 scannerContext.incrementBatchProgress(results.size()); 573 for (ExtendedCell cell : (List<ExtendedCell>) results) { 574 scannerContext.incrementSizeProgress(PrivateCellUtil.estimatedSerializedSizeOf(cell), 575 cell.heapSize()); 576 } 577 } 578 579 if (isEmptyRow || ret == FilterWrapper.FilterRowRetCode.EXCLUDE || filterRow()) { 580 incrementCountOfRowsFilteredMetric(scannerContext); 581 results.clear(); 582 boolean moreRows = nextRow(scannerContext, current); 583 if (!moreRows) { 584 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 585 } 586 587 // This row was totally filtered out, if this is NOT the last row, 588 // we should continue on. Otherwise, nothing else to do. 589 if (!shouldStop) { 590 // Read nothing as the cells was filtered, but still need to check time limit. 591 // We also check size limit because we might have read blocks in getting to this point. 592 if (scannerContext.checkAnyLimitReached(limitScope)) { 593 return true; 594 } 595 continue; 596 } 597 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 598 } 599 600 // Ok, we are done with storeHeap for this row. 601 // Now we may need to fetch additional, non-essential data into row. 602 // These values are not needed for filter to work, so we postpone their 603 // fetch to (possibly) reduce amount of data loads from disk. 604 if (this.joinedHeap != null) { 605 boolean mayHaveData = joinedHeapMayHaveData(current); 606 if (mayHaveData) { 607 joinedContinuationRow = current; 608 populateFromJoinedHeap(results, scannerContext); 609 610 if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) { 611 return true; 612 } 613 } 614 } 615 } else { 616 // Populating from the joined heap was stopped by limits, populate some more. 617 populateFromJoinedHeap(results, scannerContext); 618 if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) { 619 return true; 620 } 621 } 622 // We may have just called populateFromJoinedMap and hit the limits. If that is 623 // the case, we need to call it again on the next next() invocation. 624 if (joinedContinuationRow != null) { 625 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues(); 626 } 627 628 // Finally, we are done with both joinedHeap and storeHeap. 629 // Double check to prevent empty rows from appearing in result. It could be 630 // the case when SingleColumnValueExcludeFilter is used. 631 if (results.isEmpty()) { 632 incrementCountOfRowsFilteredMetric(scannerContext); 633 boolean moreRows = nextRow(scannerContext, current); 634 if (!moreRows) { 635 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 636 } 637 if (!shouldStop) { 638 // We check size limit because we might have read blocks in the nextRow call above, or 639 // in the call populateResults call. Only scans with hasFilterRow should reach this point, 640 // and for those scans which filter row _cells_ this is the only place we can actually 641 // enforce that the scan does not exceed limits since it bypasses all other checks above. 642 if (scannerContext.checkSizeLimit(limitScope)) { 643 return true; 644 } 645 continue; 646 } 647 } 648 649 if (shouldStop) { 650 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues(); 651 } else { 652 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues(); 653 } 654 } 655 } 656 657 private void incrementCountOfRowsFilteredMetric(ScannerContext scannerContext) { 658 region.filteredReadRequestsCount.increment(); 659 if (region.getMetrics() != null) { 660 region.getMetrics().updateFilteredRecords(); 661 } 662 663 if (scannerContext == null || !scannerContext.isTrackingMetrics()) { 664 return; 665 } 666 667 scannerContext.getMetrics() 668 .addToCounter(ServerSideScanMetrics.COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME, 1); 669 } 670 671 private void incrementCountOfRowsScannedMetric(ScannerContext scannerContext) { 672 if (scannerContext == null || !scannerContext.isTrackingMetrics()) { 673 return; 674 } 675 676 scannerContext.getMetrics() 677 .addToCounter(ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, 1); 678 } 679 680 /** Returns true when the joined heap may have data for the current row */ 681 private boolean joinedHeapMayHaveData(ExtendedCell currentRowCell) throws IOException { 682 Cell nextJoinedKv = joinedHeap.peek(); 683 boolean matchCurrentRow = 684 nextJoinedKv != null && CellUtil.matchingRows(nextJoinedKv, currentRowCell); 685 boolean matchAfterSeek = false; 686 687 // If the next value in the joined heap does not match the current row, try to seek to the 688 // correct row 689 if (!matchCurrentRow) { 690 ExtendedCell firstOnCurrentRow = PrivateCellUtil.createFirstOnRow(currentRowCell); 691 boolean seekSuccessful = this.joinedHeap.requestSeek(firstOnCurrentRow, true, true); 692 matchAfterSeek = seekSuccessful && joinedHeap.peek() != null 693 && CellUtil.matchingRows(joinedHeap.peek(), currentRowCell); 694 } 695 696 return matchCurrentRow || matchAfterSeek; 697 } 698 699 /** 700 * This function is to maintain backward compatibility for 0.94 filters. HBASE-6429 combines both 701 * filterRow & filterRow({@code List<KeyValue> kvs}) functions. While 0.94 code or older, it may 702 * not implement hasFilterRow as HBase-6429 expects because 0.94 hasFilterRow() only returns true 703 * when filterRow({@code List<KeyValue> kvs}) is overridden not the filterRow(). Therefore, the 704 * filterRow() will be skipped. 705 */ 706 private boolean filterRow() throws IOException { 707 // when hasFilterRow returns true, filter.filterRow() will be called automatically inside 708 // filterRowCells(List<Cell> kvs) so we skip that scenario here. 709 return filter != null && (!filter.hasFilterRow()) && filter.filterRow(); 710 } 711 712 private boolean filterRowKey(Cell current) throws IOException { 713 return filter != null && filter.filterRowKey(current); 714 } 715 716 /** 717 * A mocked list implementation - discards all updates. 718 */ 719 private static final List<Cell> MOCKED_LIST = new AbstractList<Cell>() { 720 721 @Override 722 public void add(int index, Cell element) { 723 // do nothing 724 } 725 726 @Override 727 public boolean addAll(int index, Collection<? extends Cell> c) { 728 return false; // this list is never changed as a result of an update 729 } 730 731 @Override 732 public KeyValue get(int index) { 733 throw new UnsupportedOperationException(); 734 } 735 736 @Override 737 public int size() { 738 return 0; 739 } 740 }; 741 742 protected boolean nextRow(ScannerContext scannerContext, Cell curRowCell) throws IOException { 743 assert this.joinedContinuationRow == null : "Trying to go to next row during joinedHeap read."; 744 745 // Enable skipping row mode, which disables limits and skips tracking progress for all 746 // but block size. We keep tracking block size because skipping a row in this way 747 // might involve reading blocks along the way. 748 scannerContext.setSkippingRow(true); 749 750 Cell next; 751 while ((next = this.storeHeap.peek()) != null && CellUtil.matchingRows(next, curRowCell)) { 752 // Check for thread interrupt status in case we have been signaled from 753 // #interruptRegionOperation. 754 region.checkInterrupt(); 755 this.storeHeap.next(MOCKED_LIST, scannerContext); 756 } 757 758 scannerContext.setSkippingRow(false); 759 resetFilters(); 760 761 // Calling the hook in CP which allows it to do a fast forward 762 return this.region.getCoprocessorHost() == null 763 || this.region.getCoprocessorHost().postScannerFilterRow(this, curRowCell); 764 } 765 766 /** 767 * Fast-path alternative to {@link #nextRow} used when the filter has provided a seek hint via 768 * {@link org.apache.hadoop.hbase.filter.Filter#getHintForRejectedRow(Cell)}. Instead of iterating 769 * through every cell in the rejected row one-by-one, this method issues a single seek to jump 770 * directly to the filter's suggested position ({@code requestSeek} for forward scans, 771 * {@code backwardSeek} for reversed scans). 772 * <p> 773 * The skipping-row mode flag is set around the seek so that block-level size tracking continues 774 * to function (consistent with {@link #nextRow}), and the filter state is reset afterwards so the 775 * next row starts with a clean filter context. 776 * <p> 777 * <strong>Stop-row invariant:</strong> This method does not validate that {@code hint} falls 778 * within the scan's stop row. If the hint overshoots, the next iteration's 779 * {@link #shouldStop(Cell)} check catches it and returns NO_MORE_VALUES. One wasted seek may 780 * occur, but correctness is maintained. 781 * <p> 782 * <strong>Metrics note:</strong> The rows-scanned metric is incremented once by the caller for 783 * the rejected row. Rows physically skipped by the seek are not individually counted — this 784 * reflects the fact that no per-row work was done for those rows. 785 * <p> 786 * <strong>Coprocessor note:</strong> {@code postScannerFilterRow} is invoked once with 787 * {@code curRowCell}, not once per skipped row. Coprocessors counting filtered rows should be 788 * aware of this semantic when the hint path is used. 789 * @param scannerContext scanner context used for limit tracking 790 * @param curRowCell the first cell of the row that was rejected by {@code filterRowKey}; 791 * passed to the coprocessor hook for observability 792 * @param hint the validated {@link ExtendedCell} returned by the filter; the scanner 793 * will seek to this position 794 * @return {@code true} if scanning should continue, {@code false} if a coprocessor requests an 795 * early stop (mirrors the contract of {@link #nextRow}) 796 * @throws IOException if the seek or the coprocessor hook signals a failure 797 */ 798 private boolean nextRowViaHint(ScannerContext scannerContext, Cell curRowCell, ExtendedCell hint) 799 throws IOException { 800 assert this.joinedContinuationRow == null : "Trying to go to next row during joinedHeap read."; 801 802 int difference = comparator.compareRows(hint, curRowCell); 803 if ((!reversed && difference > 0) || (reversed && difference < 0)) { 804 scannerContext.setSkippingRow(true); 805 if (reversed) { 806 // ReversedKeyValueHeap does not support requestSeek; use backwardSeek 807 // to position at-or-before the hint within the target row. 808 // seekToPreviousRow would skip past the hint row entirely. 809 this.storeHeap.backwardSeek(hint); 810 } else { 811 this.storeHeap.requestSeek(hint, true, true); 812 } 813 scannerContext.setSkippingRow(false); 814 815 resetFilters(); 816 817 return this.region.getCoprocessorHost() == null 818 || this.region.getCoprocessorHost().postScannerFilterRow(this, curRowCell); 819 } 820 821 return nextRow(scannerContext, curRowCell); 822 } 823 824 /** 825 * Asks the current {@link org.apache.hadoop.hbase.filter.FilterWrapper} for a seek hint to use 826 * after a row has been rejected by {@link #filterRowKey}. If the wrapped filter overrides 827 * {@link org.apache.hadoop.hbase.filter.Filter#getHintForRejectedRow(Cell)}, this returns its 828 * answer as an {@link ExtendedCell}; otherwise returns {@code null}. 829 * <p> 830 * The returned cell is validated to be an {@link ExtendedCell} because filters run on the server 831 * side and the scanner infrastructure requires {@code ExtendedCell} references. 832 * @param rowCell the first cell of the rejected row (same cell passed to {@code filterRowKey}) 833 * @return a validated {@link ExtendedCell} seek target, or {@code null} if the filter provides no 834 * hint 835 * @throws DoNotRetryIOException if the filter returns a non-{@link ExtendedCell} instance 836 * @throws IOException if the filter signals an I/O failure 837 */ 838 private ExtendedCell getHintForRejectedRow(Cell rowCell) throws IOException { 839 if (filter == null) { 840 return null; 841 } 842 Cell hint = filter.getHintForRejectedRow(rowCell); 843 if (hint == null) { 844 return null; 845 } 846 if (!(hint instanceof ExtendedCell)) { 847 throw new DoNotRetryIOException( 848 "Incorrect filter implementation: the Cell returned by getHintForRejectedRow " 849 + "is not an ExtendedCell. Filter class: " + filter.getClass().getName()); 850 } 851 return (ExtendedCell) hint; 852 } 853 854 protected boolean shouldStop(Cell currentRowCell) { 855 if (currentRowCell == null) { 856 return true; 857 } 858 if (stopRow == null || Bytes.equals(stopRow, HConstants.EMPTY_END_ROW)) { 859 return false; 860 } 861 int c = comparator.compareRows(currentRowCell, stopRow, 0, stopRow.length); 862 return c > 0 || (c == 0 && !includeStopRow); 863 } 864 865 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "IS2_INCONSISTENT_SYNC", 866 justification = "this method is only called inside close which is synchronized") 867 private void closeInternal() { 868 if (storeHeap != null) { 869 storeHeap.close(); 870 filesRead.addAll(storeHeap.getFilesRead()); 871 storeHeap = null; 872 } 873 if (joinedHeap != null) { 874 joinedHeap.close(); 875 filesRead.addAll(joinedHeap.getFilesRead()); 876 joinedHeap = null; 877 } 878 // no need to synchronize here. 879 scannerReadPoints.remove(this); 880 this.filterClosed = true; 881 } 882 883 @Override 884 public synchronized void close() { 885 TraceUtil.trace(this::closeInternal, () -> region.createRegionSpan("RegionScanner.close")); 886 } 887 888 /** 889 * Returns the set of store file paths that were successfully read by this scanner. Populated at 890 * close from the underlying store heap and joined heap (if any). 891 */ 892 @Override 893 public Set<Path> getFilesRead() { 894 return Collections.unmodifiableSet(filesRead); 895 } 896 897 @Override 898 public synchronized boolean reseek(byte[] row) throws IOException { 899 return TraceUtil.trace(() -> { 900 if (row == null) { 901 throw new IllegalArgumentException("Row cannot be null."); 902 } 903 boolean result = false; 904 region.startRegionOperation(); 905 ExtendedCell kv = PrivateCellUtil.createFirstOnRow(row, 0, (short) row.length); 906 try { 907 // use request seek to make use of the lazy seek option. See HBASE-5520 908 result = this.storeHeap.requestSeek(kv, true, true); 909 if (this.joinedHeap != null) { 910 result = this.joinedHeap.requestSeek(kv, true, true) || result; 911 } 912 } finally { 913 region.closeRegionOperation(); 914 } 915 return result; 916 }, () -> region.createRegionSpan("RegionScanner.reseek")); 917 } 918 919 @Override 920 public void shipped() throws IOException { 921 if (storeHeap != null) { 922 storeHeap.shipped(); 923 } 924 if (joinedHeap != null) { 925 joinedHeap.shipped(); 926 } 927 } 928 929 @Override 930 public void run() throws IOException { 931 // This is the RPC callback method executed. We do the close in of the scanner in this 932 // callback 933 this.close(); 934 } 935}