1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.client;
19
20 import java.io.IOException;
21 import java.io.InterruptedIOException;
22 import java.util.Arrays;
23 import java.util.LinkedList;
24 import java.util.concurrent.ExecutorService;
25
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.conf.Configuration;
29 import org.apache.hadoop.hbase.Cell;
30 import org.apache.hadoop.hbase.CellComparator;
31 import org.apache.hadoop.hbase.CellUtil;
32 import org.apache.hadoop.hbase.DoNotRetryIOException;
33 import org.apache.hadoop.hbase.HBaseConfiguration;
34 import org.apache.hadoop.hbase.HConstants;
35 import org.apache.hadoop.hbase.HRegionInfo;
36 import org.apache.hadoop.hbase.KeyValue.MetaComparator;
37 import org.apache.hadoop.hbase.NotServingRegionException;
38 import org.apache.hadoop.hbase.TableName;
39 import org.apache.hadoop.hbase.UnknownScannerException;
40 import org.apache.hadoop.hbase.classification.InterfaceAudience;
41 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
42 import org.apache.hadoop.hbase.exceptions.ScannerResetException;
43 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
44 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
45 import org.apache.hadoop.hbase.protobuf.generated.MapReduceProtos;
46 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
47 import org.apache.hadoop.hbase.util.Bytes;
48
49 import com.google.common.annotations.VisibleForTesting;
50
51
52
53
54
55 @InterfaceAudience.Private
56 public class ClientScanner extends AbstractClientScanner {
57 private static final Log LOG = LogFactory.getLog(ClientScanner.class);
58
59
60 static byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
61 protected Scan scan;
62 protected boolean closed = false;
63
64
65 protected HRegionInfo currentRegion = null;
66 protected ScannerCallableWithReplicas callable = null;
67 protected final LinkedList<Result> cache = new LinkedList<Result>();
68 protected long remainingResultSize;
69 protected int countdown;
70
71
72
73
74
75 protected final LinkedList<Result> partialResults = new LinkedList<Result>();
76
77
78
79
80
81 protected byte[] partialResultsRow = null;
82
83
84
85 protected Cell lastCellLoadedToCache = null;
86 protected final int caching;
87 protected long lastNext;
88
89 protected Result lastResult = null;
90 protected final long maxScannerResultSize;
91 private final ClusterConnection connection;
92 private final TableName tableName;
93 protected final int scannerTimeout;
94 protected boolean scanMetricsPublished = false;
95 protected RpcRetryingCaller<Result[]> caller;
96 protected RpcControllerFactory rpcControllerFactory;
97 protected Configuration conf;
98
99
100
101
102
103 protected final int primaryOperationTimeout;
104 private int retries;
105 protected final ExecutorService pool;
106 private static MetaComparator metaComparator = new MetaComparator();
107
108
109
110
111
112
113
114
115
116
117 public ClientScanner(final Configuration conf, final Scan scan, final TableName tableName,
118 ClusterConnection connection, RpcRetryingCallerFactory rpcFactory,
119 RpcControllerFactory controllerFactory, ExecutorService pool, int primaryOperationTimeout)
120 throws IOException {
121 if (LOG.isTraceEnabled()) {
122 LOG.trace(
123 "Scan table=" + tableName + ", startRow=" + Bytes.toStringBinary(scan.getStartRow()));
124 }
125 this.scan = scan;
126 this.tableName = tableName;
127 this.lastNext = System.currentTimeMillis();
128 this.connection = connection;
129 this.pool = pool;
130 this.primaryOperationTimeout = primaryOperationTimeout;
131 this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
132 HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
133 if (scan.getMaxResultSize() > 0) {
134 this.maxScannerResultSize = scan.getMaxResultSize();
135 } else {
136 this.maxScannerResultSize = conf.getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
137 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
138 }
139 this.scannerTimeout =
140 HBaseConfiguration.getInt(conf, HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
141 HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
142 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
143
144
145 initScanMetrics(scan);
146
147
148 if (this.scan.getCaching() > 0) {
149 this.caching = this.scan.getCaching();
150 } else {
151 this.caching = conf.getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING,
152 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
153 }
154
155 this.caller = rpcFactory.<Result[]> newCaller();
156 this.rpcControllerFactory = controllerFactory;
157
158 this.conf = conf;
159 initializeScannerInConstruction();
160 }
161
162 protected void initializeScannerInConstruction() throws IOException {
163
164 nextScanner(this.caching, false);
165 }
166
167 protected ClusterConnection getConnection() {
168 return this.connection;
169 }
170
171
172
173
174
175
176
177 @Deprecated
178 protected byte[] getTableName() {
179 return this.tableName.getName();
180 }
181
182 protected TableName getTable() {
183 return this.tableName;
184 }
185
186 protected int getRetries() {
187 return this.retries;
188 }
189
190 protected int getScannerTimeout() {
191 return this.scannerTimeout;
192 }
193
194 protected Configuration getConf() {
195 return this.conf;
196 }
197
198 protected Scan getScan() {
199 return scan;
200 }
201
202 protected ExecutorService getPool() {
203 return pool;
204 }
205
206 protected int getPrimaryOperationTimeout() {
207 return primaryOperationTimeout;
208 }
209
210 protected int getCaching() {
211 return caching;
212 }
213
214 protected long getTimestamp() {
215 return lastNext;
216 }
217
218 @VisibleForTesting
219 protected long getMaxResultSize() {
220 return maxScannerResultSize;
221 }
222
223
224 protected boolean checkScanStopRow(final byte[] endKey) {
225 if (this.scan.getStopRow().length > 0) {
226
227 byte[] stopRow = scan.getStopRow();
228 int cmp = Bytes.compareTo(stopRow, 0, stopRow.length, endKey, 0, endKey.length);
229 if (cmp <= 0) {
230
231
232 return true;
233 }
234 }
235 return false;
236 }
237
238 protected final void closeScanner() throws IOException {
239 if (this.callable != null) {
240 this.callable.setClose();
241 call(callable, caller, scannerTimeout);
242 this.callable = null;
243 }
244 }
245
246
247
248
249
250
251
252
253
254 protected boolean nextScanner(int nbRows, final boolean done) throws IOException {
255
256 closeScanner();
257
258
259 byte[] localStartKey;
260
261
262 if (this.currentRegion != null) {
263 byte[] endKey = this.currentRegion.getEndKey();
264 if (endKey == null || Bytes.equals(endKey, HConstants.EMPTY_BYTE_ARRAY)
265 || checkScanStopRow(endKey) || done) {
266 close();
267 if (LOG.isTraceEnabled()) {
268 LOG.trace("Finished " + this.currentRegion);
269 }
270 return false;
271 }
272 localStartKey = endKey;
273 if (LOG.isTraceEnabled()) {
274 LOG.trace("Finished " + this.currentRegion);
275 }
276 } else {
277 localStartKey = this.scan.getStartRow();
278 }
279
280 if (LOG.isDebugEnabled() && this.currentRegion != null) {
281
282 LOG.debug(
283 "Advancing internal scanner to startKey at '" + Bytes.toStringBinary(localStartKey) + "'");
284 }
285 try {
286 callable = getScannerCallable(localStartKey, nbRows);
287
288
289 call(callable, caller, scannerTimeout);
290 this.currentRegion = callable.getHRegionInfo();
291 if (this.scanMetrics != null) {
292 this.scanMetrics.countOfRegions.incrementAndGet();
293 }
294 } catch (IOException e) {
295 close();
296 throw e;
297 }
298 return true;
299 }
300
301 @VisibleForTesting
302 boolean isAnyRPCcancelled() {
303 return callable.isAnyRPCcancelled();
304 }
305
306 Result[] call(ScannerCallableWithReplicas callable,
307 RpcRetryingCaller<Result[]> caller, int scannerTimeout)
308 throws IOException, RuntimeException {
309 if (Thread.interrupted()) {
310 throw new InterruptedIOException();
311 }
312
313
314 return caller.callWithoutRetries(callable, scannerTimeout);
315 }
316
317 @InterfaceAudience.Private
318 protected ScannerCallableWithReplicas getScannerCallable(byte [] localStartKey,
319 int nbRows) {
320 scan.setStartRow(localStartKey);
321 ScannerCallable s =
322 new ScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
323 this.rpcControllerFactory);
324 s.setCaching(nbRows);
325 ScannerCallableWithReplicas sr = new ScannerCallableWithReplicas(tableName, getConnection(),
326 s, pool, primaryOperationTimeout, scan,
327 retries, scannerTimeout, caching, conf, caller);
328 return sr;
329 }
330
331
332
333
334
335
336
337
338
339
340
341
342 protected void writeScanMetrics() {
343 if (this.scanMetrics == null || scanMetricsPublished) {
344 return;
345 }
346 MapReduceProtos.ScanMetrics pScanMetrics = ProtobufUtil.toScanMetrics(scanMetrics);
347 scan.setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_DATA, pScanMetrics.toByteArray());
348 scanMetricsPublished = true;
349 }
350
351 @Override
352 public Result next() throws IOException {
353
354 if (cache.size() == 0 && this.closed) {
355 return null;
356 }
357 if (cache.size() == 0) {
358 loadCache();
359 }
360
361 if (cache.size() > 0) {
362 return cache.poll();
363 }
364
365
366 writeScanMetrics();
367 return null;
368 }
369
370 @VisibleForTesting
371 public int getCacheSize() {
372 return cache != null ? cache.size() : 0;
373 }
374
375 private boolean regionExhausted(Result[] values) {
376
377 if (values == null) {
378 return true;
379 }
380
381 if (values.length == 0 && !callable.isHeartbeatMessage()) {
382 return true;
383 }
384
385
386
387 if (callable.hasMoreResultsContext() && !callable.getServerHasMoreResults()) {
388 return true;
389 }
390 return false;
391 }
392
393 private void closeScannerIfExhausted(boolean exhausted) throws IOException {
394 if (exhausted) {
395 if (!partialResults.isEmpty()) {
396
397
398 LOG.warn("Server tells us there is no more results for this region but we still have"
399 + " partialResults, this should not happen, retry on the current scanner anyway");
400 } else {
401 closeScanner();
402 }
403 }
404 }
405
406
407
408
409 protected void loadCache() throws IOException {
410 Result[] values = null;
411 remainingResultSize = maxScannerResultSize;
412 countdown = this.caching;
413
414 if (callable == null) {
415 if (!nextScanner(countdown, false)) {
416 return;
417 }
418 }
419
420 callable.setCaching(this.caching);
421
422
423 boolean retryAfterOutOfOrderException = true;
424
425
426 int retriesLeft = getRetries();
427 for (;;) {
428 try {
429
430
431
432 values = call(callable, caller, scannerTimeout);
433
434
435
436
437
438 if (values == null && callable.switchedToADifferentReplica()) {
439
440
441 clearPartialResults();
442 this.currentRegion = callable.getHRegionInfo();
443 continue;
444 }
445 retryAfterOutOfOrderException = true;
446 } catch (DoNotRetryIOException | NeedUnmanagedConnectionException e) {
447
448
449 clearPartialResults();
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465 Throwable cause = e.getCause();
466 if ((cause != null && cause instanceof NotServingRegionException) ||
467 (cause != null && cause instanceof RegionServerStoppedException) ||
468 e instanceof OutOfOrderScannerNextException ||
469 e instanceof UnknownScannerException ||
470 e instanceof ScannerResetException) {
471
472
473 if (retriesLeft-- <= 0) {
474 throw e;
475 }
476 } else {
477 throw e;
478 }
479
480
481 if (this.lastResult != null) {
482
483
484
485
486 if (!this.lastResult.isPartial() && scan.getBatch() < 0) {
487 if (scan.isReversed()) {
488 scan.setStartRow(createClosestRowBefore(lastResult.getRow()));
489 } else {
490 scan.setStartRow(Bytes.add(lastResult.getRow(), new byte[1]));
491 }
492 } else {
493
494 scan.setStartRow(lastResult.getRow());
495 }
496 }
497 if (e instanceof OutOfOrderScannerNextException) {
498 if (retryAfterOutOfOrderException) {
499 retryAfterOutOfOrderException = false;
500 } else {
501
502 throw new DoNotRetryIOException("Failed after retry of " +
503 "OutOfOrderScannerNextException: was there a rpc timeout?", e);
504 }
505 }
506
507 this.currentRegion = null;
508
509
510 callable = null;
511
512 if (!nextScanner(countdown, false)) {
513 break;
514 }
515 continue;
516 }
517 long currentTime = System.currentTimeMillis();
518 if (this.scanMetrics != null) {
519 this.scanMetrics.sumOfMillisSecBetweenNexts.addAndGet(currentTime - lastNext);
520 }
521 lastNext = currentTime;
522
523
524
525 loadResultsToCache(values, callable.isHeartbeatMessage());
526 boolean exhausted = regionExhausted(values);
527 if (callable.isHeartbeatMessage()) {
528 if (!cache.isEmpty()) {
529
530
531
532
533 if (LOG.isTraceEnabled()) {
534 LOG.trace("Heartbeat message received and cache contains Results."
535 + " Breaking out of scan loop");
536 }
537
538
539 break;
540 }
541 }
542 if (countdown <= 0) {
543
544 closeScannerIfExhausted(exhausted);
545 break;
546 }
547 if (remainingResultSize <= 0) {
548 if (!cache.isEmpty()) {
549 closeScannerIfExhausted(exhausted);
550 break;
551 } else {
552
553
554 remainingResultSize = maxScannerResultSize;
555 }
556 }
557
558 if (exhausted) {
559 if (!partialResults.isEmpty()) {
560
561
562 LOG.warn("Server tells us there is no more results for this region but we still have"
563 + " partialResults, this should not happen, retry on the current scanner anyway");
564 continue;
565 }
566 if (!nextScanner(countdown, values == null)) {
567 break;
568 }
569 }
570 }
571 }
572
573
574
575
576
577
578
579
580
581
582
583
584
585 protected void loadResultsToCache(Result[] resultsFromServer,
586 boolean heartbeatMessage) throws IOException {
587
588 final boolean isBatchSet = scan != null && scan.getBatch() > 0;
589 final boolean allowPartials = scan != null && scan.getAllowPartialResults();
590
591
592
593
594
595
596
597 if (allowPartials || isBatchSet) {
598 addResultArrayToCache(resultsFromServer, 0,
599 (null == resultsFromServer ? 0 : resultsFromServer.length));
600 return;
601 }
602
603
604
605
606 if (resultsFromServer == null || resultsFromServer.length == 0) {
607
608
609
610 if (!partialResults.isEmpty() && !heartbeatMessage) {
611 addResultToCache(Result.createCompleteResult(partialResults));
612 clearPartialResults();
613 }
614
615 return;
616 }
617
618
619
620 Result last = resultsFromServer[resultsFromServer.length - 1];
621 Result partial = last.isPartial() ? last : null;
622
623 if (LOG.isTraceEnabled()) {
624 StringBuilder sb = new StringBuilder();
625 sb.append("number results from RPC: ").append(resultsFromServer.length).append(",");
626 sb.append("partial != null: ").append(partial != null).append(",");
627 sb.append("number of partials so far: ").append(partialResults.size());
628 LOG.trace(sb.toString());
629 }
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651 if (partial != null && partialResults.isEmpty()) {
652 addToPartialResults(partial);
653
654
655 addResultArrayToCache(resultsFromServer, 0, resultsFromServer.length - 1);
656 } else if (!partialResults.isEmpty()) {
657 for (int i = 0; i < resultsFromServer.length; i++) {
658 Result result = resultsFromServer[i];
659
660
661
662 if (Bytes.equals(partialResultsRow, result.getRow())) {
663 addToPartialResults(result);
664
665
666
667 if (!result.isPartial()) {
668 addResultToCache(Result.createCompleteResult(partialResults));
669 clearPartialResults();
670 }
671 } else {
672
673
674
675 if (!partialResults.isEmpty()) {
676 addResultToCache(Result.createCompleteResult(partialResults));
677 clearPartialResults();
678 }
679
680
681
682
683 if (result.isPartial()) {
684 addToPartialResults(result);
685 } else {
686 addResultToCache(result);
687 }
688 }
689 }
690 } else {
691 addResultArrayToCache(resultsFromServer, 0, resultsFromServer.length);
692 }
693 }
694
695
696
697
698
699
700
701 private void addResultArrayToCache(Result[] resultsToAddToCache, int start, int end) {
702 if (resultsToAddToCache != null) {
703 for (int r = start; r < end; r++) {
704 addResultToCache(resultsToAddToCache[r]);
705 }
706 }
707 }
708
709
710
711
712
713 private void addResultToCache(Result rs) {
714 rs = filterLoadedCell(rs);
715 if (rs == null) {
716 return;
717 }
718 cache.add(rs);
719 for (Cell cell : rs.rawCells()) {
720 remainingResultSize -= CellUtil.estimatedHeapSizeOf(cell);
721 }
722 countdown--;
723 this.lastResult = rs;
724 if (this.lastResult.isPartial() || scan.getBatch() > 0) {
725 updateLastCellLoadedToCache(this.lastResult);
726 } else {
727 this.lastCellLoadedToCache = null;
728 }
729 }
730
731
732
733
734
735
736
737 private void addToPartialResults(final Result result) throws IOException {
738 final byte[] row = result.getRow();
739 if (partialResultsRow != null && !Bytes.equals(row, partialResultsRow)) {
740 throw new IOException("Partial result row does not match. All partial results must come "
741 + "from the same row. partialResultsRow: " + Bytes.toString(partialResultsRow) + "row: "
742 + Bytes.toString(row));
743 }
744 partialResultsRow = row;
745 partialResults.add(result);
746 }
747
748
749
750
751 private void clearPartialResults() {
752 partialResults.clear();
753 partialResultsRow = null;
754 }
755
756 @Override
757 public void close() {
758 if (!scanMetricsPublished) writeScanMetrics();
759 if (callable != null) {
760 callable.setClose();
761 try {
762 call(callable, caller, scannerTimeout);
763 } catch (UnknownScannerException e) {
764
765
766
767 } catch (IOException e) {
768
769 LOG.warn("scanner failed to close. Exception follows: " + e);
770 }
771 callable = null;
772 }
773 closed = true;
774 }
775
776
777
778
779
780
781 protected static byte[] createClosestRowBefore(byte[] row) {
782 if (row == null) {
783 throw new IllegalArgumentException("The passed row is empty");
784 }
785 if (Bytes.equals(row, HConstants.EMPTY_BYTE_ARRAY)) {
786 return MAX_BYTE_ARRAY;
787 }
788 if (row[row.length - 1] == 0) {
789 return Arrays.copyOf(row, row.length - 1);
790 } else {
791 byte[] closestFrontRow = Arrays.copyOf(row, row.length);
792 closestFrontRow[row.length - 1] = (byte) ((closestFrontRow[row.length - 1] & 0xff) - 1);
793 closestFrontRow = Bytes.add(closestFrontRow, MAX_BYTE_ARRAY);
794 return closestFrontRow;
795 }
796 }
797
798 @Override
799 public boolean renewLease() {
800 if (callable != null) {
801
802 callable.setRenew(true);
803 try {
804 this.caller.callWithoutRetries(callable, this.scannerTimeout);
805 } catch (Exception e) {
806 return false;
807 } finally {
808 callable.setRenew(false);
809 }
810 return true;
811 }
812 return false;
813 }
814
815 protected void updateLastCellLoadedToCache(Result result) {
816 if (result.rawCells().length == 0) {
817 return;
818 }
819 this.lastCellLoadedToCache = result.rawCells()[result.rawCells().length - 1];
820 }
821
822
823
824
825
826 private int compare(Cell a, Cell b) {
827 int r = 0;
828 if (currentRegion != null && currentRegion.isMetaRegion()) {
829 r = metaComparator.compareRows(a, b);
830 } else {
831 r = CellComparator.compareRows(a, b);
832 }
833 if (r != 0) {
834 return this.scan.isReversed() ? -r : r;
835 }
836 return CellComparator.compareWithoutRow(a, b);
837 }
838
839 private Result filterLoadedCell(Result result) {
840
841
842
843
844 if (lastCellLoadedToCache == null || result.rawCells().length == 0) {
845 return result;
846 }
847 if (compare(this.lastCellLoadedToCache, result.rawCells()[0]) < 0) {
848
849
850 return result;
851 }
852 if (compare(this.lastCellLoadedToCache, result.rawCells()[result.rawCells().length - 1]) >= 0) {
853
854 return null;
855 }
856
857
858 int index = 1;
859 while (index < result.rawCells().length) {
860 if (compare(this.lastCellLoadedToCache, result.rawCells()[index]) < 0) {
861 break;
862 }
863 index++;
864 }
865 Cell[] list = Arrays.copyOfRange(result.rawCells(), index, result.rawCells().length);
866 return Result.create(list, result.getExists(), result.isStale(), result.isPartial());
867 }
868 }