View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  
20  package org.apache.hadoop.hbase.regionserver;
21  
22  import java.io.IOException;
23  import java.io.InterruptedIOException;
24  import java.util.ArrayList;
25  import java.util.List;
26  import java.util.NavigableSet;
27  import java.util.concurrent.CountDownLatch;
28  import java.util.concurrent.locks.ReentrantLock;
29  
30  import org.apache.commons.logging.Log;
31  import org.apache.commons.logging.LogFactory;
32  import org.apache.hadoop.hbase.Cell;
33  import org.apache.hadoop.hbase.CellUtil;
34  import org.apache.hadoop.hbase.DoNotRetryIOException;
35  import org.apache.hadoop.hbase.HConstants;
36  import org.apache.hadoop.hbase.KeyValue;
37  import org.apache.hadoop.hbase.KeyValue.KVComparator;
38  import org.apache.hadoop.hbase.KeyValueUtil;
39  import org.apache.hadoop.hbase.classification.InterfaceAudience;
40  import org.apache.hadoop.hbase.client.IsolationLevel;
41  import org.apache.hadoop.hbase.client.Scan;
42  import org.apache.hadoop.hbase.executor.ExecutorService;
43  import org.apache.hadoop.hbase.filter.Filter;
44  import org.apache.hadoop.hbase.regionserver.ScanQueryMatcher.MatchCode;
45  import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope;
46  import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState;
47  import org.apache.hadoop.hbase.regionserver.handler.ParallelSeekHandler;
48  import org.apache.hadoop.hbase.util.Bytes;
49  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
50  
51  import com.google.common.annotations.VisibleForTesting;
52  
53  /**
54   * Scanner scans both the memstore and the Store. Coalesce KeyValue stream
55   * into List<KeyValue> for a single row.
56   */
57  @InterfaceAudience.Private
58  public class StoreScanner extends NonReversedNonLazyKeyValueScanner
59      implements KeyValueScanner, InternalScanner, ChangedReadersObserver {
60    private static final Log LOG = LogFactory.getLog(StoreScanner.class);
61    // In unit tests, the store could be null
62    protected final Store store;
63    protected ScanQueryMatcher matcher;
64    protected KeyValueHeap heap;
65    protected boolean cacheBlocks;
66  
67    protected long countPerRow = 0;
68    protected int storeLimit = -1;
69    protected int storeOffset = 0;
70  
71    // Used to indicate that the scanner has closed (see HBASE-1107)
72    // Doesnt need to be volatile because it's always accessed via synchronized methods
73    protected boolean closing = false;
74    protected final boolean get;
75    protected final boolean explicitColumnQuery;
76    protected final boolean useRowColBloom;
77    /**
78     * A flag that enables StoreFileScanner parallel-seeking
79     */
80    protected boolean parallelSeekEnabled = false;
81    protected ExecutorService executor;
82    protected final Scan scan;
83    protected final NavigableSet<byte[]> columns;
84    protected final long oldestUnexpiredTS;
85    protected final long now;
86    protected final int minVersions;
87    protected final long maxRowSize;
88    protected final long cellsPerHeartbeatCheck;
89  
90    /**
91     * The number of KVs seen by the scanner. Includes explicitly skipped KVs, but not
92     * KVs skipped via seeking to next row/column. TODO: estimate them?
93     */
94    private long kvsScanned = 0;
95    private Cell prevCell = null;
96  
97    /** We don't ever expect to change this, the constant is just for clarity. */
98    static final boolean LAZY_SEEK_ENABLED_BY_DEFAULT = true;
99    public static final String STORESCANNER_PARALLEL_SEEK_ENABLE =
100       "hbase.storescanner.parallel.seek.enable";
101 
102   /** Used during unit testing to ensure that lazy seek does save seek ops */
103   protected static boolean lazySeekEnabledGlobally =
104       LAZY_SEEK_ENABLED_BY_DEFAULT;
105 
106   /**
107    * The number of cells scanned in between timeout checks. Specifying a larger value means that
108    * timeout checks will occur less frequently. Specifying a small value will lead to more frequent
109    * timeout checks.
110    */
111   public static final String HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK =
112       "hbase.cells.scanned.per.heartbeat.check";
113 
114   /**
115    * Default value of {@link #HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK}.
116    */
117   public static final long DEFAULT_HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK = 10000;
118 
119   // if heap == null and lastTop != null, you need to reseek given the key below
120   protected Cell lastTop = null;
121 
122   // A flag whether use pread for scan
123   private boolean scanUsePread = false;
124   // Indicates whether there was flush during the course of the scan
125   private volatile boolean flushed = false;
126   // generally we get one file from a flush
127   private List<StoreFile> flushedStoreFiles = new ArrayList<StoreFile>(1);
128   // The current list of scanners
129   private List<KeyValueScanner> currentScanners = new ArrayList<KeyValueScanner>();
130   // flush update lock
131   private ReentrantLock flushLock = new ReentrantLock();
132 
133   private final long readPt;
134 
135   // used by the injection framework to test race between StoreScanner construction and compaction
136   enum StoreScannerCompactionRace {
137     BEFORE_SEEK,
138     AFTER_SEEK,
139     COMPACT_COMPLETE
140   }
141 
142   /** An internal constructor. */
143   protected StoreScanner(Store store, Scan scan, final ScanInfo scanInfo,
144       final NavigableSet<byte[]> columns, long readPt, boolean cacheBlocks) {
145     this.readPt = readPt;
146     this.store = store;
147     this.cacheBlocks = cacheBlocks;
148     get = scan.isGetScan();
149     int numCol = columns == null ? 0 : columns.size();
150     explicitColumnQuery = numCol > 0;
151     this.scan = scan;
152     this.columns = columns;
153     this.now = EnvironmentEdgeManager.currentTime();
154     this.oldestUnexpiredTS = now - scanInfo.getTtl();
155     this.minVersions = scanInfo.getMinVersions();
156 
157      // We look up row-column Bloom filters for multi-column queries as part of
158      // the seek operation. However, we also look the row-column Bloom filter
159      // for multi-row (non-"get") scans because this is not done in
160      // StoreFile.passesBloomFilter(Scan, SortedSet<byte[]>).
161      this.useRowColBloom = numCol > 1 || (!get && numCol == 1);
162 
163      this.maxRowSize = scanInfo.getTableMaxRowSize();
164      this.scanUsePread = scan.isSmall()? true: scanInfo.isUsePread();
165      this.cellsPerHeartbeatCheck = scanInfo.getCellsPerTimeoutCheck();
166      // Parallel seeking is on if the config allows and more there is more than one store file.
167      if (this.store != null && this.store.getStorefilesCount() > 1) {
168        RegionServerServices rsService = ((HStore)store).getHRegion().getRegionServerServices();
169        if (rsService != null && scanInfo.isParallelSeekEnabled()) {
170          this.parallelSeekEnabled = true;
171          this.executor = rsService.getExecutorService();
172        }
173      }
174   }
175 
176   protected void addCurrentScanners(List<? extends KeyValueScanner> scanners) {
177     this.currentScanners.addAll(scanners);
178   }
179   /**
180    * Opens a scanner across memstore, snapshot, and all StoreFiles. Assumes we
181    * are not in a compaction.
182    *
183    * @param store who we scan
184    * @param scan the spec
185    * @param columns which columns we are scanning
186    * @throws IOException
187    */
188   public StoreScanner(Store store, ScanInfo scanInfo, Scan scan, final NavigableSet<byte[]> columns,
189       long readPt)
190   throws IOException {
191     this(store, scan, scanInfo, columns, readPt, scan.getCacheBlocks());
192     if (columns != null && scan.isRaw()) {
193       throw new DoNotRetryIOException("Cannot specify any column for a raw scan");
194     }
195     matcher = new ScanQueryMatcher(scan, scanInfo, columns,
196         ScanType.USER_SCAN, Long.MAX_VALUE, HConstants.LATEST_TIMESTAMP,
197         oldestUnexpiredTS, now, store.getCoprocessorHost());
198 
199     this.store.addChangedReaderObserver(this);
200 
201     try {
202       // Pass columns to try to filter out unnecessary StoreFiles.
203       List<KeyValueScanner> scanners = getScannersNoCompaction();
204 
205       // Seek all scanners to the start of the Row (or if the exact matching row
206       // key does not exist, then to the start of the next matching Row).
207       // Always check bloom filter to optimize the top row seek for delete
208       // family marker.
209       seekScanners(scanners, matcher.getStartKey(), explicitColumnQuery && lazySeekEnabledGlobally,
210         parallelSeekEnabled);
211 
212       // set storeLimit
213       this.storeLimit = scan.getMaxResultsPerColumnFamily();
214 
215       // set rowOffset
216       this.storeOffset = scan.getRowOffsetPerColumnFamily();
217       addCurrentScanners(scanners);
218       // Combine all seeked scanners with a heap
219       resetKVHeap(scanners, store.getComparator());
220     } catch (IOException e) {
221       // remove us from the HStore#changedReaderObservers here or we'll have no chance to
222       // and might cause memory leak
223       this.store.deleteChangedReaderObserver(this);
224       throw e;
225     }
226   }
227 
228   /**
229    * Used for compactions.<p>
230    *
231    * Opens a scanner across specified StoreFiles.
232    * @param store who we scan
233    * @param scan the spec
234    * @param scanners ancillary scanners
235    * @param smallestReadPoint the readPoint that we should use for tracking
236    *          versions
237    */
238   public StoreScanner(Store store, ScanInfo scanInfo, Scan scan,
239       List<? extends KeyValueScanner> scanners, ScanType scanType,
240       long smallestReadPoint, long earliestPutTs) throws IOException {
241     this(store, scanInfo, scan, scanners, scanType, smallestReadPoint, earliestPutTs, null, null);
242   }
243 
244   /**
245    * Used for compactions that drop deletes from a limited range of rows.<p>
246    *
247    * Opens a scanner across specified StoreFiles.
248    * @param store who we scan
249    * @param scan the spec
250    * @param scanners ancillary scanners
251    * @param smallestReadPoint the readPoint that we should use for tracking versions
252    * @param dropDeletesFromRow The inclusive left bound of the range; can be EMPTY_START_ROW.
253    * @param dropDeletesToRow The exclusive right bound of the range; can be EMPTY_END_ROW.
254    */
255   public StoreScanner(Store store, ScanInfo scanInfo, Scan scan,
256       List<? extends KeyValueScanner> scanners, long smallestReadPoint, long earliestPutTs,
257       byte[] dropDeletesFromRow, byte[] dropDeletesToRow) throws IOException {
258     this(store, scanInfo, scan, scanners, ScanType.COMPACT_RETAIN_DELETES, smallestReadPoint,
259         earliestPutTs, dropDeletesFromRow, dropDeletesToRow);
260   }
261 
262   private StoreScanner(Store store, ScanInfo scanInfo, Scan scan,
263       List<? extends KeyValueScanner> scanners, ScanType scanType, long smallestReadPoint,
264       long earliestPutTs, byte[] dropDeletesFromRow, byte[] dropDeletesToRow) throws IOException {
265     this(store, scan, scanInfo, null,
266       ((HStore)store).getHRegion().getReadpoint(IsolationLevel.READ_COMMITTED), false);
267     if (dropDeletesFromRow == null) {
268       matcher = new ScanQueryMatcher(scan, scanInfo, null, scanType, smallestReadPoint,
269           earliestPutTs, oldestUnexpiredTS, now, store.getCoprocessorHost());
270     } else {
271       matcher = new ScanQueryMatcher(scan, scanInfo, null, smallestReadPoint, earliestPutTs,
272           oldestUnexpiredTS, now, dropDeletesFromRow, dropDeletesToRow, store.getCoprocessorHost());
273     }
274 
275     // Filter the list of scanners using Bloom filters, time range, TTL, etc.
276     scanners = selectScannersFrom(scanners);
277 
278     // Seek all scanners to the initial key
279     seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled);
280     addCurrentScanners(scanners);
281     // Combine all seeked scanners with a heap
282     resetKVHeap(scanners, store.getComparator());
283   }
284 
285   @VisibleForTesting
286   StoreScanner(final Scan scan, ScanInfo scanInfo,
287       ScanType scanType, final NavigableSet<byte[]> columns,
288       final List<KeyValueScanner> scanners) throws IOException {
289     this(scan, scanInfo, scanType, columns, scanners,
290         HConstants.LATEST_TIMESTAMP,
291         // 0 is passed as readpoint because the test bypasses Store
292         0);
293   }
294 
295   @VisibleForTesting
296   StoreScanner(final Scan scan, ScanInfo scanInfo,
297     ScanType scanType, final NavigableSet<byte[]> columns,
298     final List<KeyValueScanner> scanners, long earliestPutTs)
299         throws IOException {
300     this(scan, scanInfo, scanType, columns, scanners, earliestPutTs,
301       // 0 is passed as readpoint because the test bypasses Store
302       0);
303   }
304 
305   private StoreScanner(final Scan scan, ScanInfo scanInfo,
306       ScanType scanType, final NavigableSet<byte[]> columns,
307       final List<KeyValueScanner> scanners, long earliestPutTs, long readPt)
308   throws IOException {
309     this(null, scan, scanInfo, columns, readPt, scan.getCacheBlocks());
310     this.matcher = new ScanQueryMatcher(scan, scanInfo, columns, scanType,
311         Long.MAX_VALUE, earliestPutTs, oldestUnexpiredTS, now, null);
312 
313     // In unit tests, the store could be null
314     if (this.store != null) {
315       this.store.addChangedReaderObserver(this);
316     }
317     // Seek all scanners to the initial key
318     seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled);
319     addCurrentScanners(scanners);
320     resetKVHeap(scanners, scanInfo.getComparator());
321   }
322 
323   /**
324    * Get a filtered list of scanners. Assumes we are not in a compaction.
325    * @return list of scanners to seek
326    */
327   protected List<KeyValueScanner> getScannersNoCompaction() throws IOException {
328     final boolean isCompaction = false;
329     boolean usePread = get || scanUsePread;
330     return selectScannersFrom(store.getScanners(cacheBlocks, get, usePread,
331         isCompaction, matcher, scan.getStartRow(), scan.getStopRow(), this.readPt));
332   }
333 
334   /**
335    * Seek the specified scanners with the given key
336    * @param scanners
337    * @param seekKey
338    * @param isLazy true if using lazy seek
339    * @param isParallelSeek true if using parallel seek
340    * @throws IOException
341    */
342   protected void seekScanners(List<? extends KeyValueScanner> scanners,
343       Cell seekKey, boolean isLazy, boolean isParallelSeek)
344       throws IOException {
345     // Seek all scanners to the start of the Row (or if the exact matching row
346     // key does not exist, then to the start of the next matching Row).
347     // Always check bloom filter to optimize the top row seek for delete
348     // family marker.
349     if (isLazy) {
350       for (KeyValueScanner scanner : scanners) {
351         scanner.requestSeek(seekKey, false, true);
352       }
353     } else {
354       if (!isParallelSeek) {
355         long totalScannersSoughtBytes = 0;
356         for (KeyValueScanner scanner : scanners) {
357           if (totalScannersSoughtBytes >= maxRowSize) {
358             throw new RowTooBigException("Max row size allowed: " + maxRowSize
359               + ", but row is bigger than that");
360           }
361           scanner.seek(seekKey);
362           Cell c = scanner.peek();
363           if (c != null) {
364             totalScannersSoughtBytes += CellUtil.estimatedSerializedSizeOf(c);
365           }
366         }
367       } else {
368         parallelSeek(scanners, seekKey);
369       }
370     }
371   }
372 
373   protected void resetKVHeap(List<? extends KeyValueScanner> scanners,
374       KVComparator comparator) throws IOException {
375     // Combine all seeked scanners with a heap
376     heap = new KeyValueHeap(scanners, comparator);
377   }
378 
379   /**
380    * Filters the given list of scanners using Bloom filter, time range, and
381    * TTL.
382    */
383   protected List<KeyValueScanner> selectScannersFrom(
384       final List<? extends KeyValueScanner> allScanners) {
385     boolean memOnly;
386     boolean filesOnly;
387     if (scan instanceof InternalScan) {
388       InternalScan iscan = (InternalScan)scan;
389       memOnly = iscan.isCheckOnlyMemStore();
390       filesOnly = iscan.isCheckOnlyStoreFiles();
391     } else {
392       memOnly = false;
393       filesOnly = false;
394     }
395 
396     List<KeyValueScanner> scanners =
397         new ArrayList<KeyValueScanner>(allScanners.size());
398 
399     // We can only exclude store files based on TTL if minVersions is set to 0.
400     // Otherwise, we might have to return KVs that have technically expired.
401     long expiredTimestampCutoff = minVersions == 0 ? oldestUnexpiredTS: Long.MIN_VALUE;
402 
403     // include only those scan files which pass all filters
404     for (KeyValueScanner kvs : allScanners) {
405       boolean isFile = kvs.isFileScanner();
406       if ((!isFile && filesOnly) || (isFile && memOnly)) {
407         continue;
408       }
409 
410       if (kvs.shouldUseScanner(scan, store, expiredTimestampCutoff)) {
411         scanners.add(kvs);
412       } else {
413         kvs.close();
414       }
415     }
416     return scanners;
417   }
418 
419   @Override
420   public Cell peek() {
421     if (this.heap == null) {
422       return this.lastTop;
423     }
424     return this.heap.peek();
425   }
426 
427   @Override
428   public KeyValue next() {
429     // throw runtime exception perhaps?
430     throw new RuntimeException("Never call StoreScanner.next()");
431   }
432 
433   @Override
434   public void close() {
435     if (this.closing) return;
436     this.closing = true;
437     // Under test, we dont have a this.store
438     if (this.store != null)
439       this.store.deleteChangedReaderObserver(this);
440     if (this.heap != null)
441       this.heap.close();
442     this.heap = null; // CLOSED!
443     this.lastTop = null; // If both are null, we are closed.
444   }
445 
446   @Override
447   public boolean seek(Cell key) throws IOException {
448     boolean flushed = checkFlushed();
449     // reset matcher state, in case that underlying store changed
450     checkReseek(flushed);
451     return this.heap.seek(key);
452   }
453 
454   @Override
455   public boolean next(List<Cell> outResult) throws IOException {
456     return next(outResult, NoLimitScannerContext.getInstance());
457   }
458 
459   /**
460    * Get the next row of values from this Store.
461    * @param outResult
462    * @param scannerContext
463    * @return true if there are more rows, false if scanner is done
464    */
465   @Override
466   public boolean next(List<Cell> outResult, ScannerContext scannerContext) throws IOException {
467     if (scannerContext == null) {
468       throw new IllegalArgumentException("Scanner context cannot be null");
469     }
470     boolean flushed = checkFlushed();
471     if (checkReseek(flushed)) {
472       return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
473     }
474 
475     // if the heap was left null, then the scanners had previously run out anyways, close and
476     // return.
477     if (this.heap == null) {
478       close();
479       return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
480     }
481 
482     Cell cell = this.heap.peek();
483     if (cell == null) {
484       close();
485       return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
486     }
487 
488     // only call setRow if the row changes; avoids confusing the query matcher
489     // if scanning intra-row
490     byte[] row = cell.getRowArray();
491     int offset = cell.getRowOffset();
492     short length = cell.getRowLength();
493 
494     // If no limits exists in the scope LimitScope.Between_Cells then we are sure we are changing
495     // rows. Else it is possible we are still traversing the same row so we must perform the row
496     // comparison.
497     if (!scannerContext.hasAnyLimit(LimitScope.BETWEEN_CELLS) || matcher.row == null) {
498       this.countPerRow = 0;
499       matcher.setRow(row, offset, length);
500     }
501 
502     // Clear progress away unless invoker has indicated it should be kept.
503     if (!scannerContext.getKeepProgress()) scannerContext.clearProgress();
504 
505     // Only do a sanity-check if store and comparator are available.
506     KeyValue.KVComparator comparator =
507         store != null ? store.getComparator() : null;
508 
509     int count = 0;
510     long totalBytesRead = 0;
511 
512     LOOP: do {
513       // Update and check the time limit based on the configured value of cellsPerTimeoutCheck
514       if ((kvsScanned % cellsPerHeartbeatCheck == 0)) {
515         scannerContext.updateTimeProgress();
516         if (scannerContext.checkTimeLimit(LimitScope.BETWEEN_CELLS)) {
517           return scannerContext.setScannerState(NextState.TIME_LIMIT_REACHED).hasMoreValues();
518         }
519       }
520 
521       if (prevCell != cell) ++kvsScanned; // Do object compare - we set prevKV from the same heap.
522       checkScanOrder(prevCell, cell, comparator);
523       prevCell = cell;
524 
525       ScanQueryMatcher.MatchCode qcode = matcher.match(cell);
526       qcode = optimize(qcode, cell);
527       switch(qcode) {
528         case INCLUDE:
529         case INCLUDE_AND_SEEK_NEXT_ROW:
530         case INCLUDE_AND_SEEK_NEXT_COL:
531 
532           Filter f = matcher.getFilter();
533           if (f != null) {
534             // TODO convert Scan Query Matcher to be Cell instead of KV based ?
535             cell = f.transformCell(cell);
536           }
537 
538           this.countPerRow++;
539           if (storeLimit > -1 &&
540               this.countPerRow > (storeLimit + storeOffset)) {
541             // do what SEEK_NEXT_ROW does.
542             if (!matcher.moreRowsMayExistAfter(cell)) {
543               return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
544             }
545             // Setting the matcher.row = null, will mean that after the subsequent seekToNextRow()
546             // the heap.peek() will any way be in the next row. So the SQM.match(cell) need do
547             // another compareRow to say the current row is DONE
548             matcher.row = null;
549             seekToNextRow(cell);
550             break LOOP;
551           }
552 
553           // add to results only if we have skipped #storeOffset kvs
554           // also update metric accordingly
555           if (this.countPerRow > storeOffset) {
556             outResult.add(cell);
557 
558             // Update local tracking information
559             count++;
560             totalBytesRead += CellUtil.estimatedSerializedSizeOf(cell);
561 
562             // Update the progress of the scanner context
563             scannerContext.incrementSizeProgress(CellUtil.estimatedHeapSizeOfWithoutTags(cell));
564             scannerContext.incrementBatchProgress(1);
565 
566             if (totalBytesRead > maxRowSize) {
567               throw new RowTooBigException("Max row size allowed: " + maxRowSize
568                   + ", but the row is bigger than that.");
569             }
570           }
571 
572           if (qcode == ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_ROW) {
573             if (!matcher.moreRowsMayExistAfter(cell)) {
574               return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
575             }
576             // Setting the matcher.row = null, will mean that after the subsequent seekToNextRow()
577             // the heap.peek() will any way be in the next row. So the SQM.match(cell) need do
578             // another compareRow to say the current row is DONE
579             matcher.row = null;
580             seekToNextRow(cell);
581           } else if (qcode == ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_COL) {
582             seekAsDirection(matcher.getKeyForNextColumn(cell));
583           } else {
584             this.heap.next();
585           }
586 
587           if (scannerContext.checkBatchLimit(LimitScope.BETWEEN_CELLS)) {
588             break LOOP;
589           }
590           if (scannerContext.checkSizeLimit(LimitScope.BETWEEN_CELLS)) {
591             break LOOP;
592           }
593           continue;
594 
595         case DONE:
596           // Optimization for Gets! If DONE, no more to get on this row, early exit!
597           if (this.scan.isGetScan()) {
598             // Then no more to this row... exit.
599             close();// Do all cleanup except heap.close()
600             return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
601           }
602           // We are sure that this row is done and we are in the next row.
603           // So subsequent StoresScanner.next() call need not do another compare
604           // and set the matcher.row
605           matcher.row = null;
606           return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
607 
608         case DONE_SCAN:
609           close();
610           return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
611 
612         case SEEK_NEXT_ROW:
613           // This is just a relatively simple end of scan fix, to short-cut end
614           // us if there is an endKey in the scan.
615           if (!matcher.moreRowsMayExistAfter(cell)) {
616             return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
617           }
618           // Setting the matcher.row = null, will mean that after the subsequent seekToNextRow()
619           // the heap.peek() will any way be in the next row. So the SQM.match(cell) need do
620           // another compareRow to say the current row is DONE
621           matcher.row = null;
622           seekToNextRow(cell);
623           break;
624 
625         case SEEK_NEXT_COL:
626           seekAsDirection(matcher.getKeyForNextColumn(cell));
627           break;
628 
629         case SKIP:
630           this.heap.next();
631           break;
632 
633         case SEEK_NEXT_USING_HINT:
634           // TODO convert resee to Cell?
635           Cell nextKV = matcher.getNextKeyHint(cell);
636           if (nextKV != null) {
637             seekAsDirection(nextKV);
638           } else {
639             heap.next();
640           }
641           break;
642 
643         default:
644           throw new RuntimeException("UNEXPECTED");
645       }
646     } while((cell = this.heap.peek()) != null);
647 
648     if (count > 0) {
649       return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
650     }
651 
652     // No more keys
653     close();
654     return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
655   }
656 
657   /**
658    * See if we should actually SEEK or rather just SKIP to the next Cell (see HBASE-13109).
659    * This method works together with ColumnTrackers and Filters. ColumnTrackers may issue SEEK
660    * hints, such as seek to next column, next row, or seek to an arbitrary seek key.
661    * This method intercepts these qcodes and decides whether a seek is the most efficient _actual_
662    * way to get us to the requested cell (SEEKs are more expensive than SKIP, SKIP, SKIP inside the
663    * current, loaded block).
664    * It does this by looking at the next indexed key of the current HFile. This key
665    * is then compared with the _SEEK_ key, where a SEEK key is an artificial 'last possible key
666    * on the row' (only in here, we avoid actually creating a SEEK key; in the compare we work with
667    * the current Cell but compare as though it were a seek key; see down in
668    * matcher.compareKeyForNextRow, etc). If the compare gets us onto the
669    * next block we *_SEEK, otherwise we just INCLUDE or SKIP, and let the ColumnTrackers or Filters
670    * go through the next Cell, and so on)
671    *
672    * <p>The ColumnTrackers and Filters must behave correctly in all cases, i.e. if they are past the
673    * Cells they care about they must issues a SKIP or SEEK.
674    *
675    * <p>Other notes:
676    * <ul>
677    * <li>Rows can straddle block boundaries</li>
678    * <li>Versions of columns can straddle block boundaries (i.e. column C1 at T1 might be in a
679    * different block than column C1 at T2)</li>
680    * <li>We want to SKIP and INCLUDE if the chance is high that we'll find the desired Cell after a
681    * few SKIPs...</li>
682    * <li>We want to INCLUDE_AND_SEEK and SEEK when the chance is high that we'll be able to seek
683    * past many Cells, especially if we know we need to go to the next block.</li>
684    * </ul>
685    * <p>A good proxy (best effort) to determine whether INCLUDE/SKIP is better than SEEK is whether
686    * we'll likely end up seeking to the next block (or past the next block) to get our next column.
687    * Example:
688    * <pre>
689    * |    BLOCK 1              |     BLOCK 2                   |
690    * |  r1/c1, r1/c2, r1/c3    |    r1/c4, r1/c5, r2/c1        |
691    *                                   ^         ^
692    *                                   |         |
693    *                           Next Index Key   SEEK_NEXT_ROW (before r2/c1)
694    *
695    *
696    * |    BLOCK 1                       |     BLOCK 2                      |
697    * |  r1/c1/t5, r1/c1/t4, r1/c1/t3    |    r1/c1/t2, r1/c1/T1, r1/c2/T3  |
698    *                                            ^              ^
699    *                                            |              |
700    *                                    Next Index Key        SEEK_NEXT_COL
701    * </pre>
702    * Now imagine we want columns c1 and c3 (see first diagram above), the 'Next Index Key' of r1/c4
703    * is > r1/c3 so we should seek to get to the c1 on the next row, r2. In second case, say we only
704    * want one version of c1, after we have it, a SEEK_COL will be issued to get to c2. Looking at
705    * the 'Next Index Key', it would land us in the next block, so we should SEEK. In other scenarios
706    * where the SEEK will not land us in the next block, it is very likely better to issues a series
707    * of SKIPs.
708    */
709   @VisibleForTesting
710   protected ScanQueryMatcher.MatchCode optimize(ScanQueryMatcher.MatchCode qcode, Cell cell) {
711     switch(qcode) {
712     case INCLUDE_AND_SEEK_NEXT_COL:
713     case SEEK_NEXT_COL:
714     {
715       Cell nextIndexedKey = getNextIndexedKey();
716       if (nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY
717           && matcher.compareKeyForNextColumn(nextIndexedKey, cell) >= 0) {
718         return qcode == MatchCode.SEEK_NEXT_COL ? MatchCode.SKIP : MatchCode.INCLUDE;
719       }
720       break;
721     }
722     case INCLUDE_AND_SEEK_NEXT_ROW:
723     case SEEK_NEXT_ROW:
724     {
725       // If it is a Get Scan, then we know that we are done with this row; there are no more
726       // rows beyond the current one: don't try to optimize. We are DONE. Return the *_NEXT_ROW
727       // qcode as is. When the caller gets these flags on a Get Scan, it knows it can shut down the
728       // Scan.
729       if (!this.scan.isGetScan()) {
730         Cell nextIndexedKey = getNextIndexedKey();
731         if (nextIndexedKey != null && nextIndexedKey != KeyValueScanner.NO_NEXT_INDEXED_KEY
732             && matcher.compareKeyForNextRow(nextIndexedKey, cell) > 0) {
733           return qcode == MatchCode.SEEK_NEXT_ROW ? MatchCode.SKIP : MatchCode.INCLUDE;
734         }
735       }
736       break;
737     }
738     default:
739       break;
740     }
741     return qcode;
742   }
743 
744   // Implementation of ChangedReadersObserver
745   @Override
746   public void updateReaders(List<StoreFile> sfs) throws IOException {
747     flushed = true;
748     flushLock.lock();
749     try {
750       flushedStoreFiles.addAll(sfs);
751     } finally {
752       flushLock.unlock();
753     }
754   }
755 
756   // Implementation of ChangedReadersObserver
757   protected void nullifyCurrentHeap() throws IOException {
758     if (this.closing) return;
759 
760     // All public synchronized API calls will call 'checkReseek' which will cause
761     // the scanner stack to reseek if this.heap==null && this.lastTop != null.
762     // But if two calls to updateReaders() happen without a 'next' or 'peek' then we
763     // will end up calling this.peek() which would cause a reseek in the middle of a updateReaders
764     // which is NOT what we want, not to mention could cause an NPE. So we early out here.
765     if (this.heap == null) return;
766 
767     // this could be null.
768     this.lastTop = this.heap.peek();
769 
770     //DebugPrint.println("SS updateReaders, topKey = " + lastTop);
771 
772     // close scanners to old obsolete Store files
773     this.heap.close(); // bubble thru and close all scanners.
774     this.heap = null; // the re-seeks could be slow (access HDFS) free up memory ASAP
775 
776     // Let the next() call handle re-creating and seeking
777   }
778 
779   /**
780    * @param flushed indicates if there was a flush
781    * @return true if top of heap has changed (and KeyValueHeap has to try the
782    *         next KV)
783    * @throws IOException
784    */
785   protected boolean checkReseek(boolean flushed) throws IOException {
786     if (flushed && this.lastTop != null) {
787       resetScannerStack(this.lastTop);
788       if (this.heap.peek() == null
789           || store.getComparator().compareRows(this.lastTop, this.heap.peek()) != 0) {
790         LOG.debug("Storescanner.peek() is changed where before = "
791             + this.lastTop.toString() + ",and after = " + this.heap.peek());
792         this.lastTop = null;
793         return true;
794       }
795       this.lastTop = null; // gone!
796     }
797     // else dont need to reseek
798     return false;
799   }
800 
801   protected void resetScannerStack(Cell lastTopKey) throws IOException {
802     /* When we have the scan object, should we not pass it to getScanners()
803      * to get a limited set of scanners? We did so in the constructor and we
804      * could have done it now by storing the scan object from the constructor
805      */
806 
807     final boolean isCompaction = false;
808     boolean usePread = get || scanUsePread;
809     List<KeyValueScanner> scanners = null;
810     try {
811       flushLock.lock();
812       scanners = selectScannersFrom(store.getScanners(flushedStoreFiles, cacheBlocks, get, usePread,
813         isCompaction, matcher, scan.getStartRow(), scan.getStopRow(), this.readPt, true));
814       // Clear the current set of flushed store files so that they don't get added again
815       flushedStoreFiles.clear();
816     } finally {
817       flushLock.unlock();
818     }
819 
820     // Seek the new scanners to the last key
821     seekScanners(scanners, lastTopKey, false, parallelSeekEnabled);
822     // remove the older memstore scanner
823     for (int i = 0; i < currentScanners.size(); i++) {
824       if (!currentScanners.get(i).isFileScanner()) {
825         currentScanners.remove(i);
826         break;
827       }
828     }
829     // add the newly created scanners on the flushed files and the current active memstore scanner
830     addCurrentScanners(scanners);
831     // Combine all seeked scanners with a heap
832     resetKVHeap(this.currentScanners, store.getComparator());
833     // Reset the state of the Query Matcher and set to top row.
834     // Only reset and call setRow if the row changes; avoids confusing the
835     // query matcher if scanning intra-row.
836     Cell kv = heap.peek();
837     if (kv == null) {
838       kv = lastTopKey;
839     }
840     byte[] row = kv.getRowArray();
841     int offset = kv.getRowOffset();
842     short length = kv.getRowLength();
843     if ((matcher.row == null) || !Bytes.equals(row, offset, length, matcher.row,
844         matcher.rowOffset, matcher.rowLength)) {
845       this.countPerRow = 0;
846       matcher.reset();
847       matcher.setRow(row, offset, length);
848     }
849   }
850 
851   /**
852    * Check whether scan as expected order
853    * @param prevKV
854    * @param kv
855    * @param comparator
856    * @throws IOException
857    */
858   protected void checkScanOrder(Cell prevKV, Cell kv,
859       KeyValue.KVComparator comparator) throws IOException {
860     // Check that the heap gives us KVs in an increasing order.
861     assert prevKV == null || comparator == null
862         || comparator.compare(prevKV, kv) <= 0 : "Key " + prevKV
863         + " followed by a " + "smaller key " + kv + " in cf " + store;
864   }
865 
866   protected boolean seekToNextRow(Cell kv) throws IOException {
867     return reseek(KeyValueUtil.createLastOnRow(kv));
868   }
869 
870   /**
871    * Do a reseek in a normal StoreScanner(scan forward)
872    * @param kv
873    * @return true if scanner has values left, false if end of scanner
874    * @throws IOException
875    */
876   protected boolean seekAsDirection(Cell kv)
877       throws IOException {
878     return reseek(kv);
879   }
880 
881   @Override
882   public boolean reseek(Cell kv) throws IOException {
883     boolean flushed = checkFlushed();
884     // Heap will not be null, if this is called from next() which.
885     // If called from RegionScanner.reseek(...) make sure the scanner
886     // stack is reset if needed.
887     checkReseek(flushed);
888     if (explicitColumnQuery && lazySeekEnabledGlobally) {
889       return heap.requestSeek(kv, true, useRowColBloom);
890     }
891     return heap.reseek(kv);
892   }
893 
894   protected boolean checkFlushed() {
895     // check the var without any lock. Suppose even if we see the old
896     // value here still it is ok to continue because we will not be resetting
897     // the heap but will continue with the referenced memstore's snapshot. For compactions
898     // any way we don't need the updateReaders at all to happen as we still continue with
899     // the older files
900     if (flushed) {
901       // If there is a flush and the current scan is notified on the flush ensure that the
902       // scan's heap gets reset and we do a seek on the newly flushed file.
903       if(!this.closing) {
904         this.lastTop = this.peek();
905       } else {
906         return false;
907       }
908       // reset the flag
909       flushed = false;
910       return true;
911     }
912     return false;
913   }
914 
915   @Override
916   public long getSequenceID() {
917     return 0;
918   }
919 
920   /**
921    * Seek storefiles in parallel to optimize IO latency as much as possible
922    * @param scanners the list {@link KeyValueScanner}s to be read from
923    * @param kv the KeyValue on which the operation is being requested
924    * @throws IOException
925    */
926   private void parallelSeek(final List<? extends KeyValueScanner>
927       scanners, final Cell kv) throws IOException {
928     if (scanners.isEmpty()) return;
929     int storeFileScannerCount = scanners.size();
930     CountDownLatch latch = new CountDownLatch(storeFileScannerCount);
931     List<ParallelSeekHandler> handlers =
932         new ArrayList<ParallelSeekHandler>(storeFileScannerCount);
933     for (KeyValueScanner scanner : scanners) {
934       if (scanner instanceof StoreFileScanner) {
935         ParallelSeekHandler seekHandler = new ParallelSeekHandler(scanner, kv,
936           this.readPt, latch);
937         executor.submit(seekHandler);
938         handlers.add(seekHandler);
939       } else {
940         scanner.seek(kv);
941         latch.countDown();
942       }
943     }
944 
945     try {
946       latch.await();
947     } catch (InterruptedException ie) {
948       throw (InterruptedIOException)new InterruptedIOException().initCause(ie);
949     }
950 
951     for (ParallelSeekHandler handler : handlers) {
952       if (handler.getErr() != null) {
953         throw new IOException(handler.getErr());
954       }
955     }
956   }
957 
958   /**
959    * Used in testing.
960    * @return all scanners in no particular order
961    */
962   List<KeyValueScanner> getAllScannersForTesting() {
963     List<KeyValueScanner> allScanners = new ArrayList<KeyValueScanner>();
964     KeyValueScanner current = heap.getCurrentForTesting();
965     if (current != null)
966       allScanners.add(current);
967     for (KeyValueScanner scanner : heap.getHeap())
968       allScanners.add(scanner);
969     return allScanners;
970   }
971 
972   static void enableLazySeekGlobally(boolean enable) {
973     lazySeekEnabledGlobally = enable;
974   }
975 
976   /**
977    * @return The estimated number of KVs seen by this scanner (includes some skipped KVs).
978    */
979   public long getEstimatedNumberOfKvsScanned() {
980     return this.kvsScanned;
981   }
982 
983   @Override
984   public Cell getNextIndexedKey() {
985     return this.heap.getNextIndexedKey();
986   }
987 }
988