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.ArrayList;
23 import java.util.Arrays;
24 import java.util.LinkedList;
25 import java.util.List;
26 import java.util.concurrent.ExecutorService;
27
28 import org.apache.commons.logging.Log;
29 import org.apache.commons.logging.LogFactory;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.hbase.Cell;
32 import org.apache.hadoop.hbase.CellComparator;
33 import org.apache.hadoop.hbase.CellUtil;
34 import org.apache.hadoop.hbase.DoNotRetryIOException;
35 import org.apache.hadoop.hbase.HBaseConfiguration;
36 import org.apache.hadoop.hbase.HConstants;
37 import org.apache.hadoop.hbase.HRegionInfo;
38 import org.apache.hadoop.hbase.KeyValue.MetaComparator;
39 import org.apache.hadoop.hbase.NotServingRegionException;
40 import org.apache.hadoop.hbase.TableName;
41 import org.apache.hadoop.hbase.UnknownScannerException;
42 import org.apache.hadoop.hbase.classification.InterfaceAudience;
43 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
44 import org.apache.hadoop.hbase.exceptions.ScannerResetException;
45 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
46 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
47 import org.apache.hadoop.hbase.protobuf.generated.MapReduceProtos;
48 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
49 import org.apache.hadoop.hbase.util.Bytes;
50
51 import com.google.common.annotations.VisibleForTesting;
52
53
54
55
56
57 @InterfaceAudience.Private
58 public class ClientScanner extends AbstractClientScanner {
59 private static final Log LOG = LogFactory.getLog(ClientScanner.class);
60
61
62 static byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
63 protected Scan scan;
64 protected boolean closed = false;
65
66
67 protected HRegionInfo currentRegion = null;
68 protected ScannerCallableWithReplicas callable = null;
69 protected final LinkedList<Result> cache = new LinkedList<Result>();
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 long remainingResultSize = maxScannerResultSize;
412 int 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 List<Result> resultsToAddToCache =
526 getResultsToAddToCache(values, callable.isHeartbeatMessage());
527 if (!resultsToAddToCache.isEmpty()) {
528 for (Result rs : resultsToAddToCache) {
529 rs = filterLoadedCell(rs);
530 if (rs == null) {
531 continue;
532 }
533 cache.add(rs);
534 for (Cell cell : rs.rawCells()) {
535 remainingResultSize -= CellUtil.estimatedHeapSizeOf(cell);
536 }
537 countdown--;
538 this.lastResult = rs;
539 if (this.lastResult.isPartial() || scan.getBatch() > 0) {
540 updateLastCellLoadedToCache(this.lastResult);
541 } else {
542 this.lastCellLoadedToCache = null;
543 }
544 }
545 }
546 boolean exhausted = regionExhausted(values);
547 if (callable.isHeartbeatMessage()) {
548 if (!cache.isEmpty()) {
549
550
551
552
553 if (LOG.isTraceEnabled()) {
554 LOG.trace("Heartbeat message received and cache contains Results."
555 + " Breaking out of scan loop");
556 }
557
558
559 break;
560 }
561 }
562 if (countdown <= 0) {
563
564 closeScannerIfExhausted(exhausted);
565 break;
566 }
567 if (remainingResultSize <= 0) {
568 if (!cache.isEmpty()) {
569 closeScannerIfExhausted(exhausted);
570 break;
571 } else {
572
573
574 remainingResultSize = maxScannerResultSize;
575 }
576 }
577
578 if (exhausted) {
579 if (!partialResults.isEmpty()) {
580
581
582 LOG.warn("Server tells us there is no more results for this region but we still have"
583 + " partialResults, this should not happen, retry on the current scanner anyway");
584 continue;
585 }
586 if (!nextScanner(countdown, values == null)) {
587 break;
588 }
589 }
590 }
591 }
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606 protected List<Result> getResultsToAddToCache(Result[] resultsFromServer,
607 boolean heartbeatMessage) throws IOException {
608 int resultSize = resultsFromServer != null ? resultsFromServer.length : 0;
609 List<Result> resultsToAddToCache = new ArrayList<Result>(resultSize);
610
611 final boolean isBatchSet = scan != null && scan.getBatch() > 0;
612 final boolean allowPartials = scan != null && scan.getAllowPartialResults();
613
614
615
616
617
618
619
620 if (allowPartials || isBatchSet) {
621 addResultsToList(resultsToAddToCache, resultsFromServer, 0,
622 (null == resultsFromServer ? 0 : resultsFromServer.length));
623 return resultsToAddToCache;
624 }
625
626
627
628
629 if (resultsFromServer == null || resultsFromServer.length == 0) {
630
631
632
633 if (!partialResults.isEmpty() && !heartbeatMessage) {
634 resultsToAddToCache.add(Result.createCompleteResult(partialResults));
635 clearPartialResults();
636 }
637
638 return resultsToAddToCache;
639 }
640
641
642
643 Result last = resultsFromServer[resultsFromServer.length - 1];
644 Result partial = last.isPartial() ? last : null;
645
646 if (LOG.isTraceEnabled()) {
647 StringBuilder sb = new StringBuilder();
648 sb.append("number results from RPC: ").append(resultsFromServer.length).append(",");
649 sb.append("partial != null: ").append(partial != null).append(",");
650 sb.append("number of partials so far: ").append(partialResults.size());
651 LOG.trace(sb.toString());
652 }
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674 if (partial != null && partialResults.isEmpty()) {
675 addToPartialResults(partial);
676
677
678 addResultsToList(resultsToAddToCache, resultsFromServer, 0, resultsFromServer.length - 1);
679 } else if (!partialResults.isEmpty()) {
680 for (int i = 0; i < resultsFromServer.length; i++) {
681 Result result = resultsFromServer[i];
682
683
684
685 if (Bytes.equals(partialResultsRow, result.getRow())) {
686 addToPartialResults(result);
687
688
689
690 if (!result.isPartial()) {
691 resultsToAddToCache.add(Result.createCompleteResult(partialResults));
692 clearPartialResults();
693 }
694 } else {
695
696
697
698 if (!partialResults.isEmpty()) {
699 resultsToAddToCache.add(Result.createCompleteResult(partialResults));
700 clearPartialResults();
701 }
702
703
704
705
706 if (result.isPartial()) {
707 addToPartialResults(result);
708 } else {
709 resultsToAddToCache.add(result);
710 }
711 }
712 }
713 } else {
714 addResultsToList(resultsToAddToCache, resultsFromServer, 0, resultsFromServer.length);
715 }
716
717 return resultsToAddToCache;
718 }
719
720
721
722
723
724
725
726 private void addToPartialResults(final Result result) throws IOException {
727 final byte[] row = result.getRow();
728 if (partialResultsRow != null && !Bytes.equals(row, partialResultsRow)) {
729 throw new IOException("Partial result row does not match. All partial results must come "
730 + "from the same row. partialResultsRow: " + Bytes.toString(partialResultsRow) + "row: "
731 + Bytes.toString(row));
732 }
733 partialResultsRow = row;
734 partialResults.add(result);
735 }
736
737
738
739
740 private void clearPartialResults() {
741 partialResults.clear();
742 partialResultsRow = null;
743 }
744
745
746
747
748
749
750
751
752 private void addResultsToList(List<Result> outputList, Result[] inputArray, int start, int end) {
753 if (inputArray == null || start < 0 || end > inputArray.length) return;
754
755 for (int i = start; i < end; i++) {
756 outputList.add(inputArray[i]);
757 }
758 }
759
760 @Override
761 public void close() {
762 if (!scanMetricsPublished) writeScanMetrics();
763 if (callable != null) {
764 callable.setClose();
765 try {
766 call(callable, caller, scannerTimeout);
767 } catch (UnknownScannerException e) {
768
769
770
771 } catch (IOException e) {
772
773 LOG.warn("scanner failed to close. Exception follows: " + e);
774 }
775 callable = null;
776 }
777 closed = true;
778 }
779
780
781
782
783
784
785 protected static byte[] createClosestRowBefore(byte[] row) {
786 if (row == null) {
787 throw new IllegalArgumentException("The passed row is empty");
788 }
789 if (Bytes.equals(row, HConstants.EMPTY_BYTE_ARRAY)) {
790 return MAX_BYTE_ARRAY;
791 }
792 if (row[row.length - 1] == 0) {
793 return Arrays.copyOf(row, row.length - 1);
794 } else {
795 byte[] closestFrontRow = Arrays.copyOf(row, row.length);
796 closestFrontRow[row.length - 1] = (byte) ((closestFrontRow[row.length - 1] & 0xff) - 1);
797 closestFrontRow = Bytes.add(closestFrontRow, MAX_BYTE_ARRAY);
798 return closestFrontRow;
799 }
800 }
801
802 @Override
803 public boolean renewLease() {
804 if (callable != null) {
805
806 callable.setRenew(true);
807 try {
808 this.caller.callWithoutRetries(callable, this.scannerTimeout);
809 } catch (Exception e) {
810 return false;
811 } finally {
812 callable.setRenew(false);
813 }
814 return true;
815 }
816 return false;
817 }
818
819 protected void updateLastCellLoadedToCache(Result result) {
820 if (result.rawCells().length == 0) {
821 return;
822 }
823 this.lastCellLoadedToCache = result.rawCells()[result.rawCells().length - 1];
824 }
825
826
827
828
829
830 private int compare(Cell a, Cell b) {
831 int r = 0;
832 if (currentRegion != null && currentRegion.isMetaRegion()) {
833 r = metaComparator.compareRows(a, b);
834 } else {
835 r = CellComparator.compareRows(a, b);
836 }
837 if (r != 0) {
838 return this.scan.isReversed() ? -r : r;
839 }
840 return CellComparator.compareWithoutRow(a, b);
841 }
842
843 private Result filterLoadedCell(Result result) {
844
845
846
847
848 if (lastCellLoadedToCache == null || result.rawCells().length == 0) {
849 return result;
850 }
851 if (compare(this.lastCellLoadedToCache, result.rawCells()[0]) < 0) {
852
853
854 return result;
855 }
856 if (compare(this.lastCellLoadedToCache, result.rawCells()[result.rawCells().length - 1]) >= 0) {
857
858 return null;
859 }
860
861
862 int index = 1;
863 while (index < result.rawCells().length) {
864 if (compare(this.lastCellLoadedToCache, result.rawCells()[index]) < 0) {
865 break;
866 }
867 index++;
868 }
869 Cell[] list = Arrays.copyOfRange(result.rawCells(), index, result.rawCells().length);
870 return Result.create(list, result.getExists(), result.isStale(), result.isPartial());
871 }
872 }