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.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
55
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
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
72
73 protected boolean closing = false;
74 protected final boolean get;
75 protected final boolean explicitColumnQuery;
76 protected final boolean useRowColBloom;
77
78
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
92
93
94 private long kvsScanned = 0;
95 private Cell prevCell = null;
96
97
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
103 protected static boolean lazySeekEnabledGlobally =
104 LAZY_SEEK_ENABLED_BY_DEFAULT;
105
106
107
108
109
110
111 public static final String HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK =
112 "hbase.cells.scanned.per.heartbeat.check";
113
114
115
116
117 public static final long DEFAULT_HBASE_CELLS_SCANNED_PER_HEARTBEAT_CHECK = 10000;
118
119
120 protected Cell lastTop = null;
121
122
123 private boolean scanUsePread = false;
124
125 private volatile boolean flushed = false;
126
127 private List<StoreFile> flushedStoreFiles = new ArrayList<StoreFile>(1);
128
129 private List<KeyValueScanner> currentScanners = new ArrayList<KeyValueScanner>();
130
131 private ReentrantLock flushLock = new ReentrantLock();
132
133 private final long readPt;
134
135
136 enum StoreScannerCompactionRace {
137 BEFORE_SEEK,
138 AFTER_SEEK,
139 COMPACT_COMPLETE
140 }
141
142
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
158
159
160
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
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
181
182
183
184
185
186
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
203 List<KeyValueScanner> scanners = getScannersNoCompaction();
204
205
206
207
208
209 seekScanners(scanners, matcher.getStartKey(), explicitColumnQuery && lazySeekEnabledGlobally,
210 parallelSeekEnabled);
211
212
213 this.storeLimit = scan.getMaxResultsPerColumnFamily();
214
215
216 this.storeOffset = scan.getRowOffsetPerColumnFamily();
217 addCurrentScanners(scanners);
218
219 resetKVHeap(scanners, store.getComparator());
220 } catch (IOException e) {
221
222
223 this.store.deleteChangedReaderObserver(this);
224 throw e;
225 }
226 }
227
228
229
230
231
232
233
234
235
236
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
246
247
248
249
250
251
252
253
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
276 scanners = selectScannersFrom(scanners);
277
278
279 seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled);
280 addCurrentScanners(scanners);
281
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
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
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
314 if (this.store != null) {
315 this.store.addChangedReaderObserver(this);
316 }
317
318 seekScanners(scanners, matcher.getStartKey(), false, parallelSeekEnabled);
319 addCurrentScanners(scanners);
320 resetKVHeap(scanners, scanInfo.getComparator());
321 }
322
323
324
325
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
336
337
338
339
340
341
342 protected void seekScanners(List<? extends KeyValueScanner> scanners,
343 Cell seekKey, boolean isLazy, boolean isParallelSeek)
344 throws IOException {
345
346
347
348
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
376 heap = new KeyValueHeap(scanners, comparator);
377 }
378
379
380
381
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
400
401 long expiredTimestampCutoff = minVersions == 0 ? oldestUnexpiredTS: Long.MIN_VALUE;
402
403
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
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
438 if (this.store != null)
439 this.store.deleteChangedReaderObserver(this);
440 if (this.heap != null)
441 this.heap.close();
442 this.heap = null;
443 this.lastTop = null;
444 }
445
446 @Override
447 public boolean seek(Cell key) throws IOException {
448 boolean flushed = checkFlushed();
449
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
461
462
463
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
476
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
489
490 byte[] row = cell.getRowArray();
491 int offset = cell.getRowOffset();
492 short length = cell.getRowLength();
493
494
495
496
497 if (!scannerContext.hasAnyLimit(LimitScope.BETWEEN_CELLS) || matcher.row == null) {
498 this.countPerRow = 0;
499 matcher.setRow(row, offset, length);
500 }
501
502
503 if (!scannerContext.getKeepProgress()) scannerContext.clearProgress();
504
505
506 KeyValue.KVComparator comparator =
507 store != null ? store.getComparator() : null;
508
509 int count = 0;
510 long totalBytesRead = 0;
511
512 LOOP: do {
513
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;
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
535 cell = f.transformCell(cell);
536 }
537
538 this.countPerRow++;
539 if (storeLimit > -1 &&
540 this.countPerRow > (storeLimit + storeOffset)) {
541
542 if (!matcher.moreRowsMayExistAfter(cell)) {
543 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
544 }
545
546
547
548 matcher.row = null;
549 seekToNextRow(cell);
550 break LOOP;
551 }
552
553
554
555 if (this.countPerRow > storeOffset) {
556 outResult.add(cell);
557
558
559 count++;
560 totalBytesRead += CellUtil.estimatedSerializedSizeOf(cell);
561
562
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
577
578
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
597 if (this.scan.isGetScan()) {
598
599 close();
600 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
601 }
602
603
604
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
614
615 if (!matcher.moreRowsMayExistAfter(cell)) {
616 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
617 }
618
619
620
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
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
653 close();
654 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
655 }
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
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
726
727
728
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
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
757 protected void nullifyCurrentHeap() throws IOException {
758 if (this.closing) return;
759
760
761
762
763
764
765 if (this.heap == null) return;
766
767
768 this.lastTop = this.heap.peek();
769
770
771
772
773 this.heap.close();
774 this.heap = null;
775
776
777 }
778
779
780
781
782
783
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;
796 }
797
798 return false;
799 }
800
801 protected void resetScannerStack(Cell lastTopKey) throws IOException {
802
803
804
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
815 flushedStoreFiles.clear();
816 } finally {
817 flushLock.unlock();
818 }
819
820
821 seekScanners(scanners, lastTopKey, false, parallelSeekEnabled);
822
823 for (int i = 0; i < currentScanners.size(); i++) {
824 if (!currentScanners.get(i).isFileScanner()) {
825 currentScanners.remove(i);
826 break;
827 }
828 }
829
830 addCurrentScanners(scanners);
831
832 resetKVHeap(this.currentScanners, store.getComparator());
833
834
835
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
853
854
855
856
857
858 protected void checkScanOrder(Cell prevKV, Cell kv,
859 KeyValue.KVComparator comparator) throws IOException {
860
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
872
873
874
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
885
886
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
896
897
898
899
900 if (flushed) {
901
902
903 if(!this.closing) {
904 this.lastTop = this.peek();
905 } else {
906 return false;
907 }
908
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
922
923
924
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
960
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
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