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}