1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
57
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
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
74
75 protected boolean closing = false;
76 protected final boolean get;
77 protected final boolean explicitColumnQuery;
78 protected final boolean useRowColBloom;
79
80
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
94
95
96 private final List<KeyValueScanner> scannersForDelayedClose = new ArrayList<>();
97
98
99
100
101 private long kvsScanned = 0;
102 private Cell prevCell = null;
103
104
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
110 protected static boolean lazySeekEnabledGlobally =
111 LAZY_SEEK_ENABLED_BY_DEFAULT;
112
113
114
115
116
117
118 public static final String HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK =
119 "hbase.cells.scanned.per.heartbeat.check";
120
121
122
123
124 public static final long DEFAULT_HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK = 10000;
125
126
127 protected Cell lastTop = null;
128
129
130 private boolean scanUsePread = false;
131
132 private volatile boolean flushed = false;
133
134
135 private final List<KeyValueScanner> flushedstoreFileScanners =
136 new ArrayList<KeyValueScanner>(1);
137
138 private final List<KeyValueScanner> memStoreScannersAfterFlush = new ArrayList<>(1);
139
140 private final List<KeyValueScanner> currentScanners = new ArrayList<KeyValueScanner>();
141
142 private ReentrantLock flushLock = new ReentrantLock();
143
144 private final long readPt;
145 private boolean topChanged = false;
146
147 enum StoreScannerCompactionRace {
148 BEFORE_SEEK,
149 AFTER_SEEK,
150 COMPACT_COMPLETE
151 }
152
153
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
169
170
171
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
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
192
193
194
195
196
197
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
214 List<KeyValueScanner> scanners = getScannersNoCompaction();
215
216
217
218
219
220 seekScanners(scanners, matcher.getStartKey(), explicitColumnQuery && lazySeekEnabledGlobally,
221 parallelSeekEnabled);
222
223
224 this.storeLimit = scan.getMaxResultsPerColumnFamily();
225
226
227 this.storeOffset = scan.getRowOffsetPerColumnFamily();
228 addCurrentScanners(scanners);
229
230 resetKVHeap(scanners, store.getComparator());
231 } catch (IOException e) {
232
233
234 this.store.deleteChangedReaderObserver(this);
235 throw e;
236 }
237 }
238
239
240
241
242
243
244
245
246
247
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
257
258
259
260
261
262
263
264
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
287 scanners = selectScannersFrom(scanners);
288
289
290 seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled);
291 addCurrentScanners(scanners);
292
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
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
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
325 if (this.store != null) {
326 this.store.addChangedReaderObserver(this);
327 }
328
329 seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled);
330 addCurrentScanners(scanners);
331 resetKVHeap(scanners, scanInfo.getComparator());
332 }
333
334
335
336
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
347
348
349
350
351
352
353 protected void seekScanners(List<? extends KeyValueScanner> scanners,
354 Cell seekKey, boolean isLazy, boolean isParallelSeek)
355 throws IOException {
356
357
358
359
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
387 heap = new KeyValueHeap(scanners, comparator);
388 }
389
390
391
392
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
411
412 long expiredTimestampCutoff = minVersions == 0 ? oldestUnexpiredTS: Long.MIN_VALUE;
413
414
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
441 throw new RuntimeException("Never call StoreScanner.next()");
442 }
443
444 @Override
445 public void close() {
446 if (this.closing) {
447 return;
448 }
449
450
451 if (this.store != null) {
452 this.store.deleteChangedReaderObserver(this);
453 }
454
455
456
457 flushLock.lock();
458 try {
459 this.closing = true;
460 clearAndClose(scannersForDelayedClose);
461 clearAndClose(memStoreScannersAfterFlush);
462
463
464
465 clearAndClose(flushedstoreFileScanners);
466 } finally {
467 flushLock.unlock();
468 }
469 if (this.heap != null)
470 this.heap.close();
471 this.heap = null;
472 this.lastTop = null;
473 }
474
475 @Override
476 public boolean seek(Cell key) throws IOException {
477 boolean flushed = checkFlushed();
478
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
490
491
492
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
505
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
518
519 byte[] row = cell.getRowArray();
520 int offset = cell.getRowOffset();
521 short length = cell.getRowLength();
522
523
524
525
526 if (!scannerContext.hasAnyLimit(LimitScope.BETWEEN_CELLS) || matcher.row == null) {
527 this.countPerRow = 0;
528 matcher.setRow(row, offset, length);
529 }
530
531
532 if (!scannerContext.getKeepProgress()) scannerContext.clearProgress();
533
534
535 KeyValue.KVComparator comparator =
536 store != null ? store.getComparator() : null;
537
538 int count = 0;
539 long totalBytesRead = 0;
540
541 LOOP: do {
542
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;
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
564 cell = f.transformCell(cell);
565 }
566
567 this.countPerRow++;
568 if (storeLimit > -1 &&
569 this.countPerRow > (storeLimit + storeOffset)) {
570
571 if (!matcher.moreRowsMayExistAfter(cell)) {
572 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
573 }
574
575
576
577 matcher.row = null;
578 seekToNextRow(cell);
579 break LOOP;
580 }
581
582
583
584 if (this.countPerRow > storeOffset) {
585 outResult.add(cell);
586
587
588 count++;
589 totalBytesRead += CellUtil.estimatedSerializedSizeOf(cell);
590
591
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
606
607
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
626 if (this.scan.isGetScan()) {
627
628 close();
629 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
630 }
631
632
633
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
643
644 if (!matcher.moreRowsMayExistAfter(cell)) {
645 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
646 }
647
648
649
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
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
694 close();
695 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
696 }
697
698
699
700
701
702
703
704
705
706
707
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
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
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
785
786
787
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
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
825
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
835
836
837
838
839
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
853 protected void nullifyCurrentHeap() throws IOException {
854 if (this.closing) return;
855
856
857
858
859
860
861 if (this.heap == null) return;
862
863
864 this.lastTop = this.heap.peek();
865
866
867
868
869 this.heap.close();
870 this.heap = null;
871
872
873 }
874
875
876
877
878
879
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;
893 }
894
895 topChanged = false;
896 return false;
897 }
898
899 protected void resetScannerStack(Cell lastTopKey) throws IOException {
900
901
902
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
916 flushedstoreFileScanners.clear();
917 memStoreScannersAfterFlush.clear();
918 } finally {
919 flushLock.unlock();
920 }
921
922
923 seekScanners(scanners, lastTopKey, false, parallelSeekEnabled);
924
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
932 addCurrentScanners(scanners);
933
934 resetKVHeap(this.currentScanners, store.getComparator());
935
936
937
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
955
956
957
958
959
960 protected void checkScanOrder(Cell prevKV, Cell kv,
961 KeyValue.KVComparator comparator) throws IOException {
962
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
974
975
976
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
987
988
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
998
999
1000
1001
1002 if (flushed) {
1003
1004
1005 if(!this.closing) {
1006 this.lastTop = this.peek();
1007 } else {
1008 return false;
1009 }
1010
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
1024
1025
1026
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
1062
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
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