1 /**
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
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.hbase.classification.InterfaceAudience;
31 import org.apache.hadoop.conf.Configuration;
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.HBaseConfiguration;
36 import org.apache.hadoop.hbase.HConstants;
37 import org.apache.hadoop.hbase.HRegionInfo;
38 import org.apache.hadoop.hbase.NotServingRegionException;
39 import org.apache.hadoop.hbase.TableName;
40 import org.apache.hadoop.hbase.UnknownScannerException;
41 import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory;
42 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
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 * Implements the scanner interface for the HBase client.
53 * If there are multiple regions in a table, this scanner will iterate
54 * through them all.
55 */
56 @InterfaceAudience.Private
57 public class ClientScanner extends AbstractClientScanner {
58 private static final Log LOG = LogFactory.getLog(ClientScanner.class);
59 // A byte array in which all elements are the max byte, and it is used to
60 // construct closest front row
61 static byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
62 protected Scan scan;
63 protected boolean closed = false;
64 // Current region scanner is against. Gets cleared if current region goes
65 // wonky: e.g. if it splits on us.
66 protected HRegionInfo currentRegion = null;
67 protected ScannerCallableWithReplicas callable = null;
68 protected final LinkedList<Result> cache = new LinkedList<Result>();
69 /**
70 * A list of partial results that have been returned from the server. This list should only
71 * contain results if this scanner does not have enough partial results to form the complete
72 * result.
73 */
74 protected final LinkedList<Result> partialResults = new LinkedList<Result>();
75 /**
76 * The row for which we are accumulating partial Results (i.e. the row of the Results stored
77 * inside partialResults). Changes to partialResultsRow and partialResults are kept in sync
78 * via the methods {@link #addToPartialResults(Result)} and {@link #clearPartialResults()}
79 */
80 protected byte[] partialResultsRow = null;
81 protected final int caching;
82 protected long lastNext;
83 // Keep lastResult returned successfully in case we have to reset scanner.
84 protected Result lastResult = null;
85 protected final long maxScannerResultSize;
86 private final ClusterConnection connection;
87 private final TableName tableName;
88 protected final int scannerTimeout;
89 protected boolean scanMetricsPublished = false;
90 protected RpcRetryingCaller<Result []> caller;
91 protected RpcControllerFactory rpcControllerFactory;
92 protected Configuration conf;
93 //The timeout on the primary. Applicable if there are multiple replicas for a region
94 //In that case, we will only wait for this much timeout on the primary before going
95 //to the replicas and trying the same scan. Note that the retries will still happen
96 //on each replica and the first successful results will be taken. A timeout of 0 is
97 //disallowed.
98 protected final int primaryOperationTimeout;
99 private int retries;
100 protected final ExecutorService pool;
101
102 /**
103 * Create a new ClientScanner for the specified table Note that the passed {@link Scan}'s start
104 * row maybe changed changed.
105 * @param conf The {@link Configuration} to use.
106 * @param scan {@link Scan} to use in this scanner
107 * @param tableName The table that we wish to scan
108 * @param connection Connection identifying the cluster
109 * @throws IOException
110 */
111 public ClientScanner(final Configuration conf, final Scan scan, final TableName tableName,
112 ClusterConnection connection, RpcRetryingCallerFactory rpcFactory,
113 RpcControllerFactory controllerFactory, ExecutorService pool, int primaryOperationTimeout)
114 throws IOException {
115 if (LOG.isTraceEnabled()) {
116 LOG.trace("Scan table=" + tableName
117 + ", startRow=" + Bytes.toStringBinary(scan.getStartRow()));
118 }
119 this.scan = scan;
120 this.tableName = tableName;
121 this.lastNext = System.currentTimeMillis();
122 this.connection = connection;
123 this.pool = pool;
124 this.primaryOperationTimeout = primaryOperationTimeout;
125 this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
126 HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
127 if (scan.getMaxResultSize() > 0) {
128 this.maxScannerResultSize = scan.getMaxResultSize();
129 } else {
130 this.maxScannerResultSize = conf.getLong(
131 HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
132 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
133 }
134 this.scannerTimeout = HBaseConfiguration.getInt(conf,
135 HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
136 HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
137 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
138
139 // check if application wants to collect scan metrics
140 initScanMetrics(scan);
141
142 // Use the caching from the Scan. If not set, use the default cache setting for this table.
143 if (this.scan.getCaching() > 0) {
144 this.caching = this.scan.getCaching();
145 } else {
146 this.caching = conf.getInt(
147 HConstants.HBASE_CLIENT_SCANNER_CACHING,
148 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
149 }
150
151 this.caller = rpcFactory.<Result[]> newCaller();
152 this.rpcControllerFactory = controllerFactory;
153
154 this.conf = conf;
155 initializeScannerInConstruction();
156 }
157
158 protected void initializeScannerInConstruction() throws IOException{
159 // initialize the scanner
160 nextScanner(this.caching, false);
161 }
162
163 protected ClusterConnection getConnection() {
164 return this.connection;
165 }
166
167 /**
168 * @return Table name
169 * @deprecated As of release 0.96
170 * (<a href="https://issues.apache.org/jira/browse/HBASE-9508">HBASE-9508</a>).
171 * This will be removed in HBase 2.0.0. Use {@link #getTable()}.
172 */
173 @Deprecated
174 protected byte [] getTableName() {
175 return this.tableName.getName();
176 }
177
178 protected TableName getTable() {
179 return this.tableName;
180 }
181
182 protected int getRetries() {
183 return this.retries;
184 }
185
186 protected int getScannerTimeout() {
187 return this.scannerTimeout;
188 }
189
190 protected Configuration getConf() {
191 return this.conf;
192 }
193
194 protected Scan getScan() {
195 return scan;
196 }
197
198 protected ExecutorService getPool() {
199 return pool;
200 }
201
202 protected int getPrimaryOperationTimeout() {
203 return primaryOperationTimeout;
204 }
205
206 protected int getCaching() {
207 return caching;
208 }
209
210 protected long getTimestamp() {
211 return lastNext;
212 }
213
214 @VisibleForTesting
215 protected long getMaxResultSize() {
216 return maxScannerResultSize;
217 }
218
219 // returns true if the passed region endKey
220 protected boolean checkScanStopRow(final byte [] endKey) {
221 if (this.scan.getStopRow().length > 0) {
222 // there is a stop row, check to see if we are past it.
223 byte [] stopRow = scan.getStopRow();
224 int cmp = Bytes.compareTo(stopRow, 0, stopRow.length,
225 endKey, 0, endKey.length);
226 if (cmp <= 0) {
227 // stopRow <= endKey (endKey is equals to or larger than stopRow)
228 // This is a stop.
229 return true;
230 }
231 }
232 return false; //unlikely.
233 }
234
235 private boolean possiblyNextScanner(int nbRows, final boolean done) throws IOException {
236 // If we have just switched replica, don't go to the next scanner yet. Rather, try
237 // the scanner operations on the new replica, from the right point in the scan
238 // Note that when we switched to a different replica we left it at a point
239 // where we just did the "openScanner" with the appropriate startrow
240 if (callable != null && callable.switchedToADifferentReplica()) return true;
241 return nextScanner(nbRows, done);
242 }
243
244 /*
245 * Gets a scanner for the next region. If this.currentRegion != null, then
246 * we will move to the endrow of this.currentRegion. Else we will get
247 * scanner at the scan.getStartRow(). We will go no further, just tidy
248 * up outstanding scanners, if <code>currentRegion != null</code> and
249 * <code>done</code> is true.
250 * @param nbRows
251 * @param done Server-side says we're done scanning.
252 */
253 protected boolean nextScanner(int nbRows, final boolean done)
254 throws IOException {
255 // Close the previous scanner if it's open
256 if (this.callable != null) {
257 this.callable.setClose();
258 call(callable, caller, scannerTimeout);
259 this.callable = null;
260 }
261
262 // Where to start the next scanner
263 byte [] localStartKey;
264
265 // if we're at end of table, close and return false to stop iterating
266 if (this.currentRegion != null) {
267 byte [] endKey = this.currentRegion.getEndKey();
268 if (endKey == null ||
269 Bytes.equals(endKey, HConstants.EMPTY_BYTE_ARRAY) ||
270 checkScanStopRow(endKey) ||
271 done) {
272 close();
273 if (LOG.isTraceEnabled()) {
274 LOG.trace("Finished " + this.currentRegion);
275 }
276 return false;
277 }
278 localStartKey = endKey;
279 if (LOG.isTraceEnabled()) {
280 LOG.trace("Finished " + this.currentRegion);
281 }
282 } else {
283 localStartKey = this.scan.getStartRow();
284 }
285
286 if (LOG.isDebugEnabled() && this.currentRegion != null) {
287 // Only worth logging if NOT first region in scan.
288 LOG.debug("Advancing internal scanner to startKey at '" +
289 Bytes.toStringBinary(localStartKey) + "'");
290 }
291 try {
292 callable = getScannerCallable(localStartKey, nbRows);
293 // Open a scanner on the region server starting at the
294 // beginning of the region
295 call(callable, caller, scannerTimeout);
296 this.currentRegion = callable.getHRegionInfo();
297 if (this.scanMetrics != null) {
298 this.scanMetrics.countOfRegions.incrementAndGet();
299 }
300 } catch (IOException e) {
301 close();
302 throw e;
303 }
304 return true;
305 }
306
307 @VisibleForTesting
308 boolean isAnyRPCcancelled() {
309 return callable.isAnyRPCcancelled();
310 }
311
312 Result[] call(ScannerCallableWithReplicas callable,
313 RpcRetryingCaller<Result[]> caller, int scannerTimeout)
314 throws IOException, RuntimeException {
315 if (Thread.interrupted()) {
316 throw new InterruptedIOException();
317 }
318 // callWithoutRetries is at this layer. Within the ScannerCallableWithReplicas,
319 // we do a callWithRetries
320 return caller.callWithoutRetries(callable, scannerTimeout);
321 }
322
323 @InterfaceAudience.Private
324 protected ScannerCallableWithReplicas getScannerCallable(byte [] localStartKey,
325 int nbRows) {
326 scan.setStartRow(localStartKey);
327 ScannerCallable s =
328 new ScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
329 this.rpcControllerFactory);
330 s.setCaching(nbRows);
331 ScannerCallableWithReplicas sr = new ScannerCallableWithReplicas(tableName, getConnection(),
332 s, pool, primaryOperationTimeout, scan,
333 retries, scannerTimeout, caching, conf, caller);
334 return sr;
335 }
336
337 /**
338 * Publish the scan metrics. For now, we use scan.setAttribute to pass the metrics back to the
339 * application or TableInputFormat.Later, we could push it to other systems. We don't use
340 * metrics framework because it doesn't support multi-instances of the same metrics on the same
341 * machine; for scan/map reduce scenarios, we will have multiple scans running at the same time.
342 *
343 * By default, scan metrics are disabled; if the application wants to collect them, this
344 * behavior can be turned on by calling calling {@link Scan#setScanMetricsEnabled(boolean)}
345 *
346 * <p>This invocation clears the scan metrics. Metrics are aggregated in the Scan instance.
347 */
348 protected void writeScanMetrics() {
349 if (this.scanMetrics == null || scanMetricsPublished) {
350 return;
351 }
352 MapReduceProtos.ScanMetrics pScanMetrics = ProtobufUtil.toScanMetrics(scanMetrics);
353 scan.setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_DATA, pScanMetrics.toByteArray());
354 scanMetricsPublished = true;
355 }
356
357 @Override
358 public Result next() throws IOException {
359 // If the scanner is closed and there's nothing left in the cache, next is a no-op.
360 if (cache.size() == 0 && this.closed) {
361 return null;
362 }
363 if (cache.size() == 0) {
364 loadCache();
365 }
366
367 if (cache.size() > 0) {
368 return cache.poll();
369 }
370
371 // if we exhausted this scanner before calling close, write out the scan metrics
372 writeScanMetrics();
373 return null;
374 }
375
376 @VisibleForTesting
377 public int getCacheSize() {
378 return cache != null ? cache.size() : 0;
379 }
380
381 /**
382 * Contact the servers to load more {@link Result}s in the cache.
383 */
384 protected void loadCache() throws IOException {
385 Result[] values = null;
386 long remainingResultSize = maxScannerResultSize;
387 int countdown = this.caching;
388
389 // We need to reset it if it's a new callable that was created
390 // with a countdown in nextScanner
391 callable.setCaching(this.caching);
392 // This flag is set when we want to skip the result returned. We do
393 // this when we reset scanner because it split under us.
394 boolean retryAfterOutOfOrderException = true;
395 // We don't expect that the server will have more results for us if
396 // it doesn't tell us otherwise. We rely on the size or count of results
397 boolean serverHasMoreResults = false;
398 do {
399 try {
400 // Server returns a null values if scanning is to stop. Else,
401 // returns an empty array if scanning is to go on and we've just
402 // exhausted current region.
403 values = call(callable, caller, scannerTimeout);
404 // When the replica switch happens, we need to do certain operations
405 // again. The callable will openScanner with the right startkey
406 // but we need to pick up from there. Bypass the rest of the loop
407 // and let the catch-up happen in the beginning of the loop as it
408 // happens for the cases where we see exceptions. Since only openScanner
409 // would have happened, values would be null
410 if (values == null && callable.switchedToADifferentReplica()) {
411 // Any accumulated partial results are no longer valid since the callable will
412 // openScanner with the correct startkey and we must pick up from there
413 clearPartialResults();
414 this.currentRegion = callable.getHRegionInfo();
415 continue;
416 }
417 retryAfterOutOfOrderException = true;
418 } catch (DoNotRetryIOException | NeedUnmanagedConnectionException e) {
419 // An exception was thrown which makes any partial results that we were collecting
420 // invalid. The scanner will need to be reset to the beginning of a row.
421 clearPartialResults();
422
423 // DNRIOEs are thrown to make us break out of retries. Some types of DNRIOEs want us
424 // to reset the scanner and come back in again.
425 if (e instanceof UnknownScannerException) {
426 long timeout = lastNext + scannerTimeout;
427 // If we are over the timeout, throw this exception to the client wrapped in
428 // a ScannerTimeoutException. Else, it's because the region moved and we used the old
429 // id against the new region server; reset the scanner.
430 if (timeout < System.currentTimeMillis()) {
431 LOG.info("For hints related to the following exception, please try taking a look at: " +
432 "https://hbase.apache.org/book.html#trouble.client.scantimeout");
433 long elapsed = System.currentTimeMillis() - lastNext;
434 ScannerTimeoutException ex =
435 new ScannerTimeoutException(elapsed + "ms passed since the last invocation, "
436 + "timeout is currently set to " + scannerTimeout);
437 ex.initCause(e);
438 throw ex;
439 }
440 } else {
441 // If exception is any but the list below throw it back to the client; else setup
442 // the scanner and retry.
443 Throwable cause = e.getCause();
444 if ((cause != null && cause instanceof NotServingRegionException) ||
445 (cause != null && cause instanceof RegionServerStoppedException) ||
446 e instanceof OutOfOrderScannerNextException) {
447 // Pass
448 // It is easier writing the if loop test as list of what is allowed rather than
449 // as a list of what is not allowed... so if in here, it means we do not throw.
450 } else {
451 throw e;
452 }
453 }
454 // Else, its signal from depths of ScannerCallable that we need to reset the scanner.
455 if (this.lastResult != null) {
456 // The region has moved. We need to open a brand new scanner at
457 // the new location.
458 // Reset the startRow to the row we've seen last so that the new
459 // scanner starts at the correct row. Otherwise we may see previously
460 // returned rows again.
461 // (ScannerCallable by now has "relocated" the correct region)
462 if (scan.isReversed()) {
463 scan.setStartRow(createClosestRowBefore(lastResult.getRow()));
464 } else {
465 scan.setStartRow(Bytes.add(lastResult.getRow(), new byte[1]));
466 }
467 }
468 if (e instanceof OutOfOrderScannerNextException) {
469 if (retryAfterOutOfOrderException) {
470 retryAfterOutOfOrderException = false;
471 } else {
472 // TODO: Why wrap this in a DNRIOE when it already is a DNRIOE?
473 throw new DoNotRetryIOException("Failed after retry of " +
474 "OutOfOrderScannerNextException: was there a rpc timeout?", e);
475 }
476 }
477 // Clear region.
478 this.currentRegion = null;
479 // Set this to zero so we don't try and do an rpc and close on remote server when
480 // the exception we got was UnknownScanner or the Server is going down.
481 callable = null;
482
483 // This continue will take us to while at end of loop where we will set up new scanner.
484 continue;
485 }
486 long currentTime = System.currentTimeMillis();
487 if (this.scanMetrics != null) {
488 this.scanMetrics.sumOfMillisSecBetweenNexts.addAndGet(currentTime - lastNext);
489 }
490 lastNext = currentTime;
491 // Groom the array of Results that we received back from the server before adding that
492 // Results to the scanner's cache. If partial results are not allowed to be seen by the
493 // caller, all book keeping will be performed within this method.
494 List<Result> resultsToAddToCache =
495 getResultsToAddToCache(values, callable.isHeartbeatMessage());
496 if (!resultsToAddToCache.isEmpty()) {
497 for (Result rs : resultsToAddToCache) {
498 cache.add(rs);
499 // We don't make Iterator here
500 for (Cell cell : rs.rawCells()) {
501 remainingResultSize -= CellUtil.estimatedHeapSizeOf(cell);
502 }
503 countdown--;
504 this.lastResult = rs;
505 }
506 }
507
508 // Caller of this method just wants a Result. If we see a heartbeat message, it means
509 // processing of the scan is taking a long time server side. Rather than continue to
510 // loop until a limit (e.g. size or caching) is reached, break out early to avoid causing
511 // unnecesary delays to the caller
512 if (callable.isHeartbeatMessage() && cache.size() > 0) {
513 if (LOG.isTraceEnabled()) {
514 LOG.trace("Heartbeat message received and cache contains Results."
515 + " Breaking out of scan loop");
516 }
517 break;
518 }
519
520 // We expect that the server won't have more results for us when we exhaust
521 // the size (bytes or count) of the results returned. If the server *does* inform us that
522 // there are more results, we want to avoid possiblyNextScanner(...). Only when we actually
523 // get results is the moreResults context valid.
524 if (null != values && values.length > 0 && callable.hasMoreResultsContext()) {
525 // Only adhere to more server results when we don't have any partialResults
526 // as it keeps the outer loop logic the same.
527 serverHasMoreResults = callable.getServerHasMoreResults() & partialResults.isEmpty();
528 }
529 // Values == null means server-side filter has determined we must STOP
530 // !partialResults.isEmpty() means that we are still accumulating partial Results for a
531 // row. We should not change scanners before we receive all the partial Results for that
532 // row.
533 } while (doneWithRegion(remainingResultSize, countdown, serverHasMoreResults)
534 && (!partialResults.isEmpty() || possiblyNextScanner(countdown, values == null)));
535 }
536
537 /**
538 * @param remainingResultSize
539 * @param remainingRows
540 * @param regionHasMoreResults
541 * @return true when the current region has been exhausted. When the current region has been
542 * exhausted, the region must be changed before scanning can continue
543 */
544 private boolean doneWithRegion(long remainingResultSize, int remainingRows,
545 boolean regionHasMoreResults) {
546 return remainingResultSize > 0 && remainingRows > 0 && !regionHasMoreResults;
547 }
548
549 /**
550 * This method ensures all of our book keeping regarding partial results is kept up to date. This
551 * method should be called once we know that the results we received back from the RPC request do
552 * not contain errors. We return a list of results that should be added to the cache. In general,
553 * this list will contain all NON-partial results from the input array (unless the client has
554 * specified that they are okay with receiving partial results)
555 * @param resultsFromServer The array of {@link Result}s returned from the server
556 * @param heartbeatMessage Flag indicating whether or not the response received from the server
557 * represented a complete response, or a heartbeat message that was sent to keep the
558 * client-server connection alive
559 * @return the list of results that should be added to the cache.
560 * @throws IOException
561 */
562 protected List<Result>
563 getResultsToAddToCache(Result[] resultsFromServer, boolean heartbeatMessage)
564 throws IOException {
565 int resultSize = resultsFromServer != null ? resultsFromServer.length : 0;
566 List<Result> resultsToAddToCache = new ArrayList<Result>(resultSize);
567
568 final boolean isBatchSet = scan != null && scan.getBatch() > 0;
569 final boolean allowPartials = scan != null && scan.getAllowPartialResults();
570
571 // If the caller has indicated in their scan that they are okay with seeing partial results,
572 // then simply add all results to the list. Note that since scan batching also returns results
573 // for a row in pieces we treat batch being set as equivalent to allowing partials. The
574 // implication of treating batching as equivalent to partial results is that it is possible
575 // the caller will receive a result back where the number of cells in the result is less than
576 // the batch size even though it may not be the last group of cells for that row.
577 if (allowPartials || isBatchSet) {
578 addResultsToList(resultsToAddToCache, resultsFromServer, 0,
579 (null == resultsFromServer ? 0 : resultsFromServer.length));
580 return resultsToAddToCache;
581 }
582
583 // If no results were returned it indicates that either we have the all the partial results
584 // necessary to construct the complete result or the server had to send a heartbeat message
585 // to the client to keep the client-server connection alive
586 if (resultsFromServer == null || resultsFromServer.length == 0) {
587 // If this response was an empty heartbeat message, then we have not exhausted the region
588 // and thus there may be more partials server side that still need to be added to the partial
589 // list before we form the complete Result
590 if (!partialResults.isEmpty() && !heartbeatMessage) {
591 resultsToAddToCache.add(Result.createCompleteResult(partialResults));
592 clearPartialResults();
593 }
594
595 return resultsToAddToCache;
596 }
597
598 // In every RPC response there should be at most a single partial result. Furthermore, if
599 // there is a partial result, it is guaranteed to be in the last position of the array.
600 Result last = resultsFromServer[resultsFromServer.length - 1];
601 Result partial = last.isPartial() ? last : null;
602
603 if (LOG.isTraceEnabled()) {
604 StringBuilder sb = new StringBuilder();
605 sb.append("number results from RPC: ").append(resultsFromServer.length).append(",");
606 sb.append("partial != null: ").append(partial != null).append(",");
607 sb.append("number of partials so far: ").append(partialResults.size());
608 LOG.trace(sb.toString());
609 }
610
611 // There are three possibilities cases that can occur while handling partial results
612 //
613 // 1. (partial != null && partialResults.isEmpty())
614 // This is the first partial result that we have received. It should be added to
615 // the list of partialResults and await the next RPC request at which point another
616 // portion of the complete result will be received
617 //
618 // 2. !partialResults.isEmpty()
619 // Since our partialResults list is not empty it means that we have been accumulating partial
620 // Results for a particular row. We cannot form the complete/whole Result for that row until
621 // all partials for the row have been received. Thus we loop through all of the Results
622 // returned from the server and determine whether or not all partial Results for the row have
623 // been received. We know that we have received all of the partial Results for the row when:
624 // i) We notice a row change in the Results
625 // ii) We see a Result for the partial row that is NOT marked as a partial Result
626 //
627 // 3. (partial == null && partialResults.isEmpty())
628 // Business as usual. We are not accumulating partial results and there wasn't a partial result
629 // in the RPC response. This means that all of the results we received from the server are
630 // complete and can be added directly to the cache
631 if (partial != null && partialResults.isEmpty()) {
632 addToPartialResults(partial);
633
634 // Exclude the last result, it's a partial
635 addResultsToList(resultsToAddToCache, resultsFromServer, 0, resultsFromServer.length - 1);
636 } else if (!partialResults.isEmpty()) {
637 for (int i = 0; i < resultsFromServer.length; i++) {
638 Result result = resultsFromServer[i];
639
640 // This result is from the same row as the partial Results. Add it to the list of partials
641 // and check if it was the last partial Result for that row
642 if (Bytes.equals(partialResultsRow, result.getRow())) {
643 addToPartialResults(result);
644
645 // If the result is not a partial, it is a signal to us that it is the last Result we
646 // need to form the complete Result client-side
647 if (!result.isPartial()) {
648 resultsToAddToCache.add(Result.createCompleteResult(partialResults));
649 clearPartialResults();
650 }
651 } else {
652 // The row of this result differs from the row of the partial results we have received so
653 // far. If our list of partials isn't empty, this is a signal to form the complete Result
654 // since the row has now changed
655 if (!partialResults.isEmpty()) {
656 resultsToAddToCache.add(Result.createCompleteResult(partialResults));
657 clearPartialResults();
658 }
659
660 // It's possible that in one response from the server we receive the final partial for
661 // one row and receive a partial for a different row. Thus, make sure that all Results
662 // are added to the proper list
663 if (result.isPartial()) {
664 addToPartialResults(result);
665 } else {
666 resultsToAddToCache.add(result);
667 }
668 }
669 }
670 } else { // partial == null && partialResults.isEmpty() -- business as usual
671 addResultsToList(resultsToAddToCache, resultsFromServer, 0, resultsFromServer.length);
672 }
673
674 return resultsToAddToCache;
675 }
676
677 /**
678 * A convenience method for adding a Result to our list of partials. This method ensure that only
679 * Results that belong to the same row as the other partials can be added to the list.
680 * @param result The result that we want to add to our list of partial Results
681 * @throws IOException
682 */
683 private void addToPartialResults(final Result result) throws IOException {
684 final byte[] row = result.getRow();
685 if (partialResultsRow != null && !Bytes.equals(row, partialResultsRow)) {
686 throw new IOException("Partial result row does not match. All partial results must come "
687 + "from the same row. partialResultsRow: " + Bytes.toString(partialResultsRow) + "row: "
688 + Bytes.toString(row));
689 }
690 partialResultsRow = row;
691 partialResults.add(result);
692 }
693
694 /**
695 * Convenience method for clearing the list of partials and resetting the partialResultsRow.
696 */
697 private void clearPartialResults() {
698 partialResults.clear();
699 partialResultsRow = null;
700 }
701
702 /**
703 * Helper method for adding results between the indices [start, end) to the outputList
704 * @param outputList the list that results will be added to
705 * @param inputArray the array that results are taken from
706 * @param start beginning index (inclusive)
707 * @param end ending index (exclusive)
708 */
709 private void addResultsToList(List<Result> outputList, Result[] inputArray, int start, int end) {
710 if (inputArray == null || start < 0 || end > inputArray.length) return;
711
712 for (int i = start; i < end; i++) {
713 outputList.add(inputArray[i]);
714 }
715 }
716
717 @Override
718 public void close() {
719 if (!scanMetricsPublished) writeScanMetrics();
720 if (callable != null) {
721 callable.setClose();
722 try {
723 call(callable, caller, scannerTimeout);
724 } catch (UnknownScannerException e) {
725 // We used to catch this error, interpret, and rethrow. However, we
726 // have since decided that it's not nice for a scanner's close to
727 // throw exceptions. Chances are it was just due to lease time out.
728 } catch (IOException e) {
729 /* An exception other than UnknownScanner is unexpected. */
730 LOG.warn("scanner failed to close. Exception follows: " + e);
731 }
732 callable = null;
733 }
734 closed = true;
735 }
736
737 /**
738 * Create the closest row before the specified row
739 * @param row
740 * @return a new byte array which is the closest front row of the specified one
741 */
742 protected static byte[] createClosestRowBefore(byte[] row) {
743 if (row == null) {
744 throw new IllegalArgumentException("The passed row is empty");
745 }
746 if (Bytes.equals(row, HConstants.EMPTY_BYTE_ARRAY)) {
747 return MAX_BYTE_ARRAY;
748 }
749 if (row[row.length - 1] == 0) {
750 return Arrays.copyOf(row, row.length - 1);
751 } else {
752 byte[] closestFrontRow = Arrays.copyOf(row, row.length);
753 closestFrontRow[row.length - 1] = (byte) ((closestFrontRow[row.length - 1] & 0xff) - 1);
754 closestFrontRow = Bytes.add(closestFrontRow, MAX_BYTE_ARRAY);
755 return closestFrontRow;
756 }
757 }
758
759 @Override
760 public boolean renewLease() {
761 if (callable != null) {
762 // do not return any rows, do not advance the scanner
763 callable.setRenew(true);
764 try {
765 this.caller.callWithoutRetries(callable, this.scannerTimeout);
766 } catch (Exception e) {
767 return false;
768 } finally {
769 callable.setRenew(false);
770 }
771 return true;
772 }
773 return false;
774 }
775 }