View Javadoc

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.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   * Implements the scanner interface for the HBase client. If there are multiple regions in a table,
53   * this scanner will iterate through them all.
54   */
55  @InterfaceAudience.Private
56  public class ClientScanner extends AbstractClientScanner {
57    private static final Log LOG = LogFactory.getLog(ClientScanner.class);
58    // A byte array in which all elements are the max byte, and it is used to
59    // construct closest front row
60    static byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
61    protected Scan scan;
62    protected boolean closed = false;
63    // Current region scanner is against. Gets cleared if current region goes
64    // wonky: e.g. if it splits on us.
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     * A list of partial results that have been returned from the server. This list should only
72     * contain results if this scanner does not have enough partial results to form the complete
73     * result.
74     */
75    protected final LinkedList<Result> partialResults = new LinkedList<Result>();
76    /**
77     * The row for which we are accumulating partial Results (i.e. the row of the Results stored
78     * inside partialResults). Changes to partialResultsRow and partialResults are kept in sync via
79     * the methods {@link #addToPartialResults(Result)} and {@link #clearPartialResults()}
80     */
81    protected byte[] partialResultsRow = null;
82    /**
83     * The last cell from a not full Row which is added to cache
84     */
85    protected Cell lastCellLoadedToCache = null;
86    protected final int caching;
87    protected long lastNext;
88    // Keep lastResult returned successfully in case we have to reset scanner.
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    // The timeout on the primary. Applicable if there are multiple replicas for a region
99    // In that case, we will only wait for this much timeout on the primary before going
100   // to the replicas and trying the same scan. Note that the retries will still happen
101   // on each replica and the first successful results will be taken. A timeout of 0 is
102   // disallowed.
103   protected final int primaryOperationTimeout;
104   private int retries;
105   protected final ExecutorService pool;
106   private static MetaComparator metaComparator = new MetaComparator();
107 
108   /**
109    * Create a new ClientScanner for the specified table Note that the passed {@link Scan}'s start
110    * row maybe changed changed.
111    * @param conf The {@link Configuration} to use.
112    * @param scan {@link Scan} to use in this scanner
113    * @param tableName The table that we wish to scan
114    * @param connection Connection identifying the cluster
115    * @throws IOException
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     // check if application wants to collect scan metrics
145     initScanMetrics(scan);
146 
147     // Use the caching from the Scan. If not set, use the default cache setting for this table.
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     // initialize the scanner
164     nextScanner(this.caching, false);
165   }
166 
167   protected ClusterConnection getConnection() {
168     return this.connection;
169   }
170 
171   /**
172    * @return Table name
173    * @deprecated As of release 0.96
174    *             (<a href="https://issues.apache.org/jira/browse/HBASE-9508">HBASE-9508</a>). This
175    *             will be removed in HBase 2.0.0. Use {@link #getTable()}.
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   // returns true if the passed region endKey
224   protected boolean checkScanStopRow(final byte[] endKey) {
225     if (this.scan.getStopRow().length > 0) {
226       // there is a stop row, check to see if we are past it.
227       byte[] stopRow = scan.getStopRow();
228       int cmp = Bytes.compareTo(stopRow, 0, stopRow.length, endKey, 0, endKey.length);
229       if (cmp <= 0) {
230         // stopRow <= endKey (endKey is equals to or larger than stopRow)
231         // This is a stop.
232         return true;
233       }
234     }
235     return false; // unlikely.
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    * Gets a scanner for the next region. If this.currentRegion != null, then we will move to the
248    * endrow of this.currentRegion. Else we will get scanner at the scan.getStartRow(). We will go no
249    * further, just tidy up outstanding scanners, if <code>currentRegion != null</code> and
250    * <code>done</code> is true.
251    * @param nbRows
252    * @param done Server-side says we're done scanning.
253    */
254   protected boolean nextScanner(int nbRows, final boolean done) throws IOException {
255     // Close the previous scanner if it's open
256     closeScanner();
257 
258     // Where to start the next scanner
259     byte[] localStartKey;
260 
261     // if we're at end of table, close and return false to stop iterating
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       // Only worth logging if NOT first region in scan.
282       LOG.debug(
283         "Advancing internal scanner to startKey at '" + Bytes.toStringBinary(localStartKey) + "'");
284     }
285     try {
286       callable = getScannerCallable(localStartKey, nbRows);
287       // Open a scanner on the region server starting at the
288       // beginning of the region
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     // callWithoutRetries is at this layer. Within the ScannerCallableWithReplicas,
313     // we do a callWithRetries
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      * Publish the scan metrics. For now, we use scan.setAttribute to pass the metrics back to the
333      * application or TableInputFormat.Later, we could push it to other systems. We don't use
334      * metrics framework because it doesn't support multi-instances of the same metrics on the same
335      * machine; for scan/map reduce scenarios, we will have multiple scans running at the same time.
336      *
337      * By default, scan metrics are disabled; if the application wants to collect them, this
338      * behavior can be turned on by calling calling {@link Scan#setScanMetricsEnabled(boolean)}
339      * 
340      * <p>This invocation clears the scan metrics. Metrics are aggregated in the Scan instance.
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       // If the scanner is closed and there's nothing left in the cache, next is a no-op.
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       // if we exhausted this scanner before calling close, write out the scan metrics
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     // This means the server tells us the whole scan operation is done. Usually decided by filter.
377     if (values == null) {
378       return true;
379     }
380     // Not a heartbeat message and we get nothing, this means the region is exhausted
381     if (values.length == 0 && !callable.isHeartbeatMessage()) {
382       return true;
383     }
384     // Server tells us that it has no more results for this region. Notice that this flag is get
385     // from the ScanResponse.getMoreResultsInRegion, not ScanResponse.getMoreResults. If the latter
386     // one is false then we will get a null values and quit in the first condition of this method.
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         // XXX: continue if there are partial results. But in fact server should not set
397         // hasMoreResults to false if there are partial results.
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    * Contact the servers to load more {@link Result}s in the cache.
408    */
409   protected void loadCache() throws IOException {
410     Result[] values = null;
411     remainingResultSize = maxScannerResultSize;
412     countdown = this.caching;
413     // This is possible if we just stopped at the boundary of a region in the previous call.
414     if (callable == null) {
415       if (!nextScanner(countdown, false)) {
416         return;
417       }
418     }
419     // We need to reset it if it's a new callable that was created with a countdown in nextScanner
420     callable.setCaching(this.caching);
421     // This flag is set when we want to skip the result returned. We do
422     // this when we reset scanner because it split under us.
423     boolean retryAfterOutOfOrderException = true;
424     // Even if we are retrying due to UnknownScannerException, ScannerResetException, etc. we should
425     // make sure that we are not retrying indefinitely.
426     int retriesLeft = getRetries();
427     for (;;) {
428       try {
429         // Server returns a null values if scanning is to stop. Else,
430         // returns an empty array if scanning is to go on and we've just
431         // exhausted current region.
432         values = call(callable, caller, scannerTimeout);
433         // When the replica switch happens, we need to do certain operations again.
434         // The callable will openScanner with the right startkey but we need to pick up
435         // from there. Bypass the rest of the loop and let the catch-up happen in the beginning
436         // of the loop as it happens for the cases where we see exceptions.
437         // Since only openScanner would have happened, values would be null
438         if (values == null && callable.switchedToADifferentReplica()) {
439           // Any accumulated partial results are no longer valid since the callable will
440           // openScanner with the correct startkey and we must pick up from there
441           clearPartialResults();
442           this.currentRegion = callable.getHRegionInfo();
443           continue;
444         }
445         retryAfterOutOfOrderException = true;
446       } catch (DoNotRetryIOException | NeedUnmanagedConnectionException e) {
447         // An exception was thrown which makes any partial results that we were collecting
448         // invalid. The scanner will need to be reset to the beginning of a row.
449         clearPartialResults();
450 
451         // Unfortunately, DNRIOE is used in two different semantics.
452         // (1) The first is to close the client scanner and bubble up the exception all the way
453         // to the application. This is preferred when the exception is really un-recoverable
454         // (like CorruptHFileException, etc). Plain DoNotRetryIOException also falls into this
455         // bucket usually.
456         // (2) Second semantics is to close the current region scanner only, but continue the
457         // client scanner by overriding the exception. This is usually UnknownScannerException,
458         // OutOfOrderScannerNextException, etc where the region scanner has to be closed, but the
459         // application-level ClientScanner has to continue without bubbling up the exception to
460         // the client. See RSRpcServices to see how it throws DNRIOE's.
461         // See also: HBASE-16604, HBASE-17187
462 
463         // If exception is any but the list below throw it back to the client; else setup
464         // the scanner and retry.
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           // Pass. It is easier writing the if loop test as list of what is allowed rather than
472           // as a list of what is not allowed... so if in here, it means we do not throw.
473           if (retriesLeft-- <= 0) {
474             throw e; // no more retries
475           }
476         } else {
477           throw e;
478         }
479 
480         // Else, its signal from depths of ScannerCallable that we need to reset the scanner.
481         if (this.lastResult != null) {
482           // The region has moved. We need to open a brand new scanner at the new location.
483           // Reset the startRow to the row we've seen last so that the new scanner starts at
484           // the correct row. Otherwise we may see previously returned rows again.
485           // (ScannerCallable by now has "relocated" the correct region)
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             // we need rescan this row because we only loaded partial row before
494             scan.setStartRow(lastResult.getRow());
495           }
496         }
497         if (e instanceof OutOfOrderScannerNextException) {
498           if (retryAfterOutOfOrderException) {
499             retryAfterOutOfOrderException = false;
500           } else {
501             // TODO: Why wrap this in a DNRIOE when it already is a DNRIOE?
502             throw new DoNotRetryIOException("Failed after retry of " +
503                 "OutOfOrderScannerNextException: was there a rpc timeout?", e);
504           }
505         }
506         // Clear region.
507         this.currentRegion = null;
508         // Set this to zero so we don't try and do an rpc and close on remote server when
509         // the exception we got was UnknownScanner or the Server is going down.
510         callable = null;
511         // reopen the scanner
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       // Groom the array of Results that we received back from the server before adding that
523       // Results to the scanner's cache. If partial results are not allowed to be seen by the
524       // caller, all book keeping will be performed within this method.
525       loadResultsToCache(values, callable.isHeartbeatMessage());
526       boolean exhausted = regionExhausted(values);
527       if (callable.isHeartbeatMessage()) {
528         if (!cache.isEmpty()) {
529           // Caller of this method just wants a Result. If we see a heartbeat message, it means
530           // processing of the scan is taking a long time server side. Rather than continue to
531           // loop until a limit (e.g. size or caching) is reached, break out early to avoid causing
532           // unnecesary delays to the caller
533           if (LOG.isTraceEnabled()) {
534             LOG.trace("Heartbeat message received and cache contains Results."
535                 + " Breaking out of scan loop");
536           }
537           // we know that the region has not been exhausted yet so just break without calling
538           // closeScannerIfExhausted
539           break;
540         }
541       }
542       if (countdown <= 0) {
543         // we have enough result.
544         closeScannerIfExhausted(exhausted);
545         break;
546       }
547       if (remainingResultSize <= 0) {
548         if (!cache.isEmpty()) {
549           closeScannerIfExhausted(exhausted);
550           break;
551         } else {
552           // we have reached the max result size but we still can not find anything to return to the
553           // user. Reset the maxResultSize and try again.
554           remainingResultSize = maxScannerResultSize;
555         }
556       }
557       // we are done with the current region
558       if (exhausted) {
559         if (!partialResults.isEmpty()) {
560           // XXX: continue if there are partial results. But in fact server should not set
561           // hasMoreResults to false if there are partial results.
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    * This method ensures all of our book keeping regarding partial results is kept up to date. This
575    * method should be called once we know that the results we received back from the RPC request do
576    * not contain errors. We groom a list of results that should be added to the cache. In general,
577    * this list will contain all NON-partial results from the input array (unless the client has
578    * specified that they are okay with receiving partial results)
579    * @param resultsFromServer The array of {@link Result}s returned from the server
580    * @param heartbeatMessage Flag indicating whether or not the response received from the server
581    *          represented a complete response, or a heartbeat message that was sent to keep the
582    *          client-server connection alive
583    * @throws IOException
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     // If the caller has indicated in their scan that they are okay with seeing partial results,
592     // then simply add all results to the cache. Note that since scan batching also returns results
593     // for a row in pieces we treat batch being set as equivalent to allowing partials. The
594     // implication of treating batching as equivalent to partial results is that it is possible
595     // the caller will receive a result back where the number of cells in the result is less than
596     // the batch size even though it may not be the last group of cells for that row.
597     if (allowPartials || isBatchSet) {
598       addResultArrayToCache(resultsFromServer, 0,
599         (null == resultsFromServer ? 0 : resultsFromServer.length));
600       return;
601     }
602 
603     // If no results were returned it indicates that either we have the all the partial results
604     // necessary to construct the complete result or the server had to send a heartbeat message
605     // to the client to keep the client-server connection alive
606     if (resultsFromServer == null || resultsFromServer.length == 0) {
607       // If this response was an empty heartbeat message, then we have not exhausted the region
608       // and thus there may be more partials server side that still need to be added to the partial
609       // list before we form the complete Result
610       if (!partialResults.isEmpty() && !heartbeatMessage) {
611         addResultToCache(Result.createCompleteResult(partialResults));
612         clearPartialResults();
613       }
614 
615       return;
616     }
617 
618     // In every RPC response there should be at most a single partial result. Furthermore, if
619     // there is a partial result, it is guaranteed to be in the last position of the array.
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     // There are three possibilities cases that can occur while handling partial results
632     //
633     // 1. (partial != null && partialResults.isEmpty())
634     // This is the first partial result that we have received. It should be added to
635     // the list of partialResults and await the next RPC request at which point another
636     // portion of the complete result will be received
637     //
638     // 2. !partialResults.isEmpty()
639     // Since our partialResults list is not empty it means that we have been accumulating partial
640     // Results for a particular row. We cannot form the complete/whole Result for that row until
641     // all partials for the row have been received. Thus we loop through all of the Results
642     // returned from the server and determine whether or not all partial Results for the row have
643     // been received. We know that we have received all of the partial Results for the row when:
644     // i) We notice a row change in the Results
645     // ii) We see a Result for the partial row that is NOT marked as a partial Result
646     //
647     // 3. (partial == null && partialResults.isEmpty())
648     // Business as usual. We are not accumulating partial results and there wasn't a partial result
649     // in the RPC response. This means that all of the results we received from the server are
650     // complete and can be added directly to the cache
651     if (partial != null && partialResults.isEmpty()) {
652       addToPartialResults(partial);
653 
654       // Exclude the last result, it's a partial
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         // This result is from the same row as the partial Results. Add it to the list of partials
661         // and check if it was the last partial Result for that row
662         if (Bytes.equals(partialResultsRow, result.getRow())) {
663           addToPartialResults(result);
664 
665           // If the result is not a partial, it is a signal to us that it is the last Result we
666           // need to form the complete Result client-side
667           if (!result.isPartial()) {
668             addResultToCache(Result.createCompleteResult(partialResults));
669             clearPartialResults();
670           }
671         } else {
672           // The row of this result differs from the row of the partial results we have received so
673           // far. If our list of partials isn't empty, this is a signal to form the complete Result
674           // since the row has now changed
675           if (!partialResults.isEmpty()) {
676             addResultToCache(Result.createCompleteResult(partialResults));
677             clearPartialResults();
678           }
679 
680           // It's possible that in one response from the server we receive the final partial for
681           // one row and receive a partial for a different row. Thus, make sure that all Results
682           // are added to the proper list
683           if (result.isPartial()) {
684             addToPartialResults(result);
685           } else {
686             addResultToCache(result);
687           }
688         }
689       }
690     } else { // partial == null && partialResults.isEmpty() -- business as usual
691       addResultArrayToCache(resultsFromServer, 0, resultsFromServer.length);
692     }
693   }
694 
695   /**
696    * Add result array received from server to cache
697    * @param resultsToAddToCache The array of Results returned from the server
698    * @param start start index to cache from Results array
699    * @param end last index to cache from Results array
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    * Add result received from server or result constructed from partials to cache
711    * @param rs Result to cache from Results array or constructed from partial results
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    * A convenience method for adding a Result to our list of partials. This method ensure that only
733    * Results that belong to the same row as the other partials can be added to the list.
734    * @param result The result that we want to add to our list of partial Results
735    * @throws IOException
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    * Convenience method for clearing the list of partials and resetting the partialResultsRow.
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            // We used to catch this error, interpret, and rethrow. However, we
765            // have since decided that it's not nice for a scanner's close to
766            // throw exceptions. Chances are it was just due to lease time out.
767         } catch (IOException e) {
768            /* An exception other than UnknownScanner is unexpected. */
769            LOG.warn("scanner failed to close. Exception follows: " + e);
770         }
771         callable = null;
772       }
773       closed = true;
774     }
775 
776   /**
777    * Create the closest row before the specified row
778    * @param row
779    * @return a new byte array which is the closest front row of the specified one
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       // do not return any rows, do not advance the scanner
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    * Compare two Cells considering reversed scanner. ReversedScanner only reverses rows, not
824    * columns.
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     // we only filter result when last result is partial
841     // so lastCellLoadedToCache and result should have same row key.
842     // However, if 1) read some cells; 1.1) delete this row at the same time 2) move region;
843     // 3) read more cell. lastCellLoadedToCache and result will be not at same row.
844     if (lastCellLoadedToCache == null || result.rawCells().length == 0) {
845       return result;
846     }
847     if (compare(this.lastCellLoadedToCache, result.rawCells()[0]) < 0) {
848       // The first cell of this result is larger than the last cell of loadcache.
849       // If user do not allow partial result, it must be true.
850       return result;
851     }
852     if (compare(this.lastCellLoadedToCache, result.rawCells()[result.rawCells().length - 1]) >= 0) {
853       // The last cell of this result is smaller than the last cell of loadcache, skip all.
854       return null;
855     }
856 
857     // The first one must not in filtered result, we start at the second.
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 }