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