hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zhang...@apache.org
Subject hbase git commit: HBASE-17740 Correct the semantic of batch and partial for async client
Date Thu, 16 Mar 2017 01:45:10 GMT
Repository: hbase
Updated Branches:
  refs/heads/master 0ecb67825 -> 1849e8a5a


HBASE-17740 Correct the semantic of batch and partial for async client


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/1849e8a5
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/1849e8a5
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/1849e8a5

Branch: refs/heads/master
Commit: 1849e8a5a77373b5fb8e354c3f20214a80eb8c1a
Parents: 0ecb678
Author: zhangduo <zhangduo@apache.org>
Authored: Wed Mar 15 18:26:51 2017 +0800
Committer: zhangduo <zhangduo@apache.org>
Committed: Thu Mar 16 09:44:23 2017 +0800

----------------------------------------------------------------------
 .../client/AllowPartialScanResultCache.java     |  31 ++-
 .../hadoop/hbase/client/AsyncClientScanner.java |   4 +-
 .../hbase/client/BatchScanResultCache.java      | 142 +++++++++++
 .../hadoop/hbase/client/ClientScanner.java      | 253 +------------------
 .../hadoop/hbase/client/ConnectionUtils.java    |  14 +
 .../org/apache/hadoop/hbase/client/Result.java  |  72 +++---
 .../client/TestAllowPartialScanResultCache.java |  33 ++-
 .../hbase/client/TestBatchScanResultCache.java  | 113 +++++++++
 .../TestCompleteResultScanResultCache.java      |   5 +-
 .../client/TestRawAsyncTablePartialScan.java    | 119 +++++++++
 10 files changed, 471 insertions(+), 315 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AllowPartialScanResultCache.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AllowPartialScanResultCache.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AllowPartialScanResultCache.java
index caecfd4..82f1ea0 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AllowPartialScanResultCache.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AllowPartialScanResultCache.java
@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hbase.client;
 
+import static org.apache.hadoop.hbase.client.ConnectionUtils.filterCells;
+
 import java.io.IOException;
 import java.util.Arrays;
 
@@ -36,10 +38,6 @@ class AllowPartialScanResultCache implements ScanResultCache {
   // beginning of a row when retry.
   private Cell lastCell;
 
-  private Result filterCells(Result result) {
-    return lastCell == null ? result : ConnectionUtils.filterCells(result, lastCell);
-  }
-
   private void updateLastCell(Result result) {
     lastCell = result.rawCells()[result.rawCells().length - 1];
   }
@@ -49,22 +47,23 @@ class AllowPartialScanResultCache implements ScanResultCache {
     if (results.length == 0) {
       return EMPTY_RESULT_ARRAY;
     }
-    Result first = filterCells(results[0]);
-    if (results.length == 1) {
-      if (first == null) {
-        // do not update last cell if we filter out all cells
-        return EMPTY_RESULT_ARRAY;
+    int i;
+    for (i = 0; i < results.length; i++) {
+      Result r = filterCells(results[i], lastCell);
+      if (r != null) {
+        results[i] = r;
+        break;
       }
-      updateLastCell(results[0]);
-      results[0] = first;
-      return results;
+    }
+    if (i == results.length) {
+      return EMPTY_RESULT_ARRAY;
     }
     updateLastCell(results[results.length - 1]);
-    if (first == null) {
-      return Arrays.copyOfRange(results, 1, results.length);
+    if (i > 0) {
+      return Arrays.copyOfRange(results, i, results.length);
+    } else {
+      return results;
     }
-    results[0] = first;
-    return results;
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
index 2215d36..fa7aa81 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClientScanner.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.hbase.client;
 
 import static org.apache.hadoop.hbase.HConstants.EMPTY_END_ROW;
 import static org.apache.hadoop.hbase.HConstants.EMPTY_START_ROW;
+import static org.apache.hadoop.hbase.client.ConnectionUtils.createScanResultCache;
 import static org.apache.hadoop.hbase.client.ConnectionUtils.getLocateType;
 
 import java.io.IOException;
@@ -86,8 +87,7 @@ class AsyncClientScanner {
     this.scanTimeoutNs = scanTimeoutNs;
     this.rpcTimeoutNs = rpcTimeoutNs;
     this.startLogErrorsCnt = startLogErrorsCnt;
-    this.resultCache = scan.getAllowPartialResults() || scan.getBatch() > 0
-        ? new AllowPartialScanResultCache() : new CompleteScanResultCache();
+    this.resultCache = createScanResultCache(scan);
   }
 
   private static final class OpenScannerResponse {

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BatchScanResultCache.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BatchScanResultCache.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BatchScanResultCache.java
new file mode 100644
index 0000000..9ab959b
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BatchScanResultCache.java
@@ -0,0 +1,142 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import static org.apache.hadoop.hbase.client.ConnectionUtils.filterCells;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * A scan result cache for batched scan, i.e,
+ * {@code scan.getBatch() > 0 && !scan.getAllowPartialResults()}.
+ * <p>
+ * If user setBatch(5) and rpc returns 3+5+5+5+3 cells, we should return 5+5+5+5+1 to user. setBatch
+ * doesn't mean setAllowPartialResult(true).
+ */
+@InterfaceAudience.Private
+public class BatchScanResultCache implements ScanResultCache {
+
+  private final int batch;
+
+  // used to filter out the cells that already returned to user as we always start from the
+  // beginning of a row when retry.
+  private Cell lastCell;
+
+  private final Deque<Result> partialResults = new ArrayDeque<>();
+
+  private int numCellsOfPartialResults;
+
+  public BatchScanResultCache(int batch) {
+    this.batch = batch;
+  }
+
+  private void updateLastCell(Result result) {
+    lastCell = result.rawCells()[result.rawCells().length - 1];
+  }
+
+  private Result createCompletedResult() throws IOException {
+    Result result = Result.createCompleteResult(partialResults);
+    partialResults.clear();
+    numCellsOfPartialResults = 0;
+    return result;
+  }
+
+  // Add new result to the partial list and return a batched Result if caching size exceed batching
+  // limit. As the RS will also respect the scan.getBatch, we can make sure that we will get only
+  // one Result back at most(or null, which means we do not have enough cells).
+  private Result regroupResults(Result result) {
+    partialResults.addLast(result);
+    numCellsOfPartialResults += result.size();
+    if (numCellsOfPartialResults < batch) {
+      return null;
+    }
+    Cell[] cells = new Cell[batch];
+    int cellCount = 0;
+    boolean stale = false;
+    for (;;) {
+      Result r = partialResults.pollFirst();
+      stale = stale || r.isStale();
+      int newCellCount = cellCount + r.size();
+      if (newCellCount > batch) {
+        // We have more cells than expected, so split the current result
+        int len = batch - cellCount;
+        System.arraycopy(r.rawCells(), 0, cells, cellCount, len);
+        Cell[] remainingCells = new Cell[r.size() - len];
+        System.arraycopy(r.rawCells(), len, remainingCells, 0, r.size() - len);
+        partialResults.addFirst(
+          Result.create(remainingCells, r.getExists(), r.isStale(), r.mayHaveMoreCellsInRow()));
+        break;
+      }
+      System.arraycopy(r.rawCells(), 0, cells, cellCount, r.size());
+      if (newCellCount == batch) {
+        break;
+      }
+      cellCount = newCellCount;
+    }
+    numCellsOfPartialResults -= batch;
+    return Result.create(cells, null, stale,
+      result.mayHaveMoreCellsInRow() || !partialResults.isEmpty());
+  }
+
+  @Override
+  public Result[] addAndGet(Result[] results, boolean isHeartbeatMessage) throws IOException {
+    if (results.length == 0) {
+      if (!partialResults.isEmpty() && !isHeartbeatMessage) {
+        return new Result[] { createCompletedResult() };
+      }
+      return EMPTY_RESULT_ARRAY;
+    }
+    List<Result> regroupedResults = new ArrayList<>();
+    for (Result result : results) {
+      result = filterCells(result, lastCell);
+      if (result == null) {
+        continue;
+      }
+      // check if we have a row change
+      if (!partialResults.isEmpty() &&
+          !Bytes.equals(partialResults.peek().getRow(), result.getRow())) {
+        regroupedResults.add(createCompletedResult());
+      }
+      Result regroupedResult = regroupResults(result);
+      if (regroupedResult != null) {
+        regroupedResults.add(regroupedResult);
+        // only update last cell when we actually return it to user.
+        updateLastCell(regroupedResult);
+      }
+      if (!result.mayHaveMoreCellsInRow() && !partialResults.isEmpty()) {
+        // We are done for this row
+        regroupedResults.add(createCompletedResult());
+      }
+    }
+    return regroupedResults.toArray(new Result[0]);
+  }
+
+  @Override
+  public void clear() {
+    partialResults.clear();
+    numCellsOfPartialResults = 0;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
index bd3d4ef..a8b029f 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java
@@ -18,16 +18,15 @@
 package org.apache.hadoop.hbase.client;
 
 import static org.apache.hadoop.hbase.client.ConnectionUtils.calcEstimatedSize;
+import static org.apache.hadoop.hbase.client.ConnectionUtils.createScanResultCache;
 import static org.apache.hadoop.hbase.client.ConnectionUtils.numberOfIndividualRows;
 
 import com.google.common.annotations.VisibleForTesting;
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.LinkedList;
-import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.ExecutorService;
 
@@ -35,8 +34,6 @@ import org.apache.commons.lang.mutable.MutableBoolean;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellComparator;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HConstants;
@@ -69,24 +66,7 @@ public abstract class ClientScanner extends AbstractClientScanner {
   protected HRegionInfo currentRegion = null;
   protected ScannerCallableWithReplicas callable = null;
   protected Queue<Result> cache;
-  /**
-   * A list of partial results that have been returned from the server. This list should only
-   * contain results if this scanner does not have enough partial results to form the complete
-   * result.
-   */
-  protected int partialResultsCellSizes = 0;
-  protected final LinkedList<Result> partialResults = new LinkedList<>();
-
-  /**
-   * The row for which we are accumulating partial Results (i.e. the row of the Results stored
-   * inside partialResults). Changes to partialResultsRow and partialResults are kept in sync via
-   * the methods {@link #regroupResults(Result)} and {@link #clearPartialResults()}
-   */
-  protected byte[] partialResultsRow = null;
-  /**
-   * The last cell from a not full Row which is added to cache
-   */
-  protected Cell lastCellLoadedToCache = null;
+  private final ScanResultCache scanResultCache;
   protected final int caching;
   protected long lastNext;
   // Keep lastResult returned successfully in case we have to reset scanner.
@@ -159,6 +139,8 @@ public abstract class ClientScanner extends AbstractClientScanner {
     this.rpcControllerFactory = controllerFactory;
 
     this.conf = conf;
+
+    this.scanResultCache = createScanResultCache(scan);
     initCache();
   }
 
@@ -356,14 +338,7 @@ public abstract class ClientScanner extends AbstractClientScanner {
 
   private void closeScannerIfExhausted(boolean exhausted) throws IOException {
     if (exhausted) {
-      if (!partialResults.isEmpty()) {
-        // XXX: continue if there are partial results. But in fact server should not set
-        // hasMoreResults to false if there are partial results.
-        LOG.warn("Server tells us there is no more results for this region but we still have" +
-            " partialResults, this should not happen, retry on the current scanner anyway");
-      } else {
-        closeScanner();
-      }
+      closeScanner();
     }
   }
 
@@ -371,7 +346,7 @@ public abstract class ClientScanner extends AbstractClientScanner {
       MutableBoolean retryAfterOutOfOrderException, int retriesLeft) throws DoNotRetryIOException {
     // An exception was thrown which makes any partial results that we were collecting
     // invalid. The scanner will need to be reset to the beginning of a row.
-    clearPartialResults();
+    scanResultCache.clear();
 
     // Unfortunately, DNRIOE is used in two different semantics.
     // (1) The first is to close the client scanner and bubble up the exception all the way
@@ -465,7 +440,7 @@ public abstract class ClientScanner extends AbstractClientScanner {
         if (callable.switchedToADifferentReplica()) {
           // Any accumulated partial results are no longer valid since the callable will
           // openScanner with the correct startkey and we must pick up from there
-          clearPartialResults();
+          scanResultCache.clear();
           this.currentRegion = callable.getHRegionInfo();
         }
         retryAfterOutOfOrderException.setValue(true);
@@ -485,29 +460,19 @@ public abstract class ClientScanner extends AbstractClientScanner {
       // Groom the array of Results that we received back from the server before adding that
       // Results to the scanner's cache. If partial results are not allowed to be seen by the
       // caller, all book keeping will be performed within this method.
-      List<Result> resultsToAddToCache =
-          getResultsToAddToCache(values, callable.isHeartbeatMessage());
-      if (!resultsToAddToCache.isEmpty()) {
+      Result[] resultsToAddToCache = scanResultCache.addAndGet(values, callable.isHeartbeatMessage());
+      if (resultsToAddToCache.length > 0) {
         for (Result rs : resultsToAddToCache) {
-          rs = filterLoadedCell(rs);
-          if (rs == null) {
-            continue;
-          }
-
           cache.add(rs);
           long estimatedHeapSizeOfResult = calcEstimatedSize(rs);
           countdown--;
           remainingResultSize -= estimatedHeapSizeOfResult;
           addEstimatedSize(estimatedHeapSizeOfResult);
           this.lastResult = rs;
-          if (this.lastResult.mayHaveMoreCellsInRow()) {
-            updateLastCellLoadedToCache(this.lastResult);
-          } else {
-            this.lastCellLoadedToCache = null;
-          }
         }
-        if (scan.getLimit() > 0 && !resultsToAddToCache.isEmpty()) {
-          int newLimit = scan.getLimit() - numberOfIndividualRows(resultsToAddToCache);
+        if (scan.getLimit() > 0) {
+          int newLimit =
+              scan.getLimit() - numberOfIndividualRows(Arrays.asList(resultsToAddToCache));
           assert newLimit >= 0;
           scan.setLimit(newLimit);
         }
@@ -550,13 +515,6 @@ public abstract class ClientScanner extends AbstractClientScanner {
       }
       // we are done with the current region
       if (regionExhausted) {
-        if (!partialResults.isEmpty()) {
-          // XXX: continue if there are partial results. But in fact server should not set
-          // hasMoreResults to false if there are partial results.
-          LOG.warn("Server tells us there is no more results for this region but we still have" +
-              " partialResults, this should not happen, retry on the current scanner anyway");
-          continue;
-        }
         if (!moveToNextRegion()) {
           break;
         }
@@ -573,142 +531,6 @@ public abstract class ClientScanner extends AbstractClientScanner {
     return cache != null ? cache.size() : 0;
   }
 
-  /**
-   * This method ensures all of our book keeping regarding partial results is kept up to date. This
-   * method should be called once we know that the results we received back from the RPC request do
-   * not contain errors. We return a list of results that should be added to the cache. In general,
-   * this list will contain all NON-partial results from the input array (unless the client has
-   * specified that they are okay with receiving partial results)
-   * @param resultsFromServer The array of {@link Result}s returned from the server
-   * @param heartbeatMessage Flag indicating whether or not the response received from the server
-   *          represented a complete response, or a heartbeat message that was sent to keep the
-   *          client-server connection alive
-   * @return the list of results that should be added to the cache.
-   * @throws IOException
-   */
-  protected List<Result> getResultsToAddToCache(Result[] resultsFromServer,
-      boolean heartbeatMessage) throws IOException {
-    int resultSize = resultsFromServer != null ? resultsFromServer.length : 0;
-    List<Result> resultsToAddToCache = new ArrayList<>(resultSize);
-
-    // If the caller has indicated in their scan that they are okay with seeing partial results,
-    // then simply add all results to the list. Note allowPartial and setBatch are not same, we can
-    // return here if allow partials and we will handle batching later.
-    if (scan.getAllowPartialResults()) {
-      addResultsToList(resultsToAddToCache, resultsFromServer, 0,
-        (null == resultsFromServer ? 0 : resultsFromServer.length));
-      return resultsToAddToCache;
-    }
-
-    // If no results were returned it indicates that either we have the all the partial results
-    // necessary to construct the complete result or the server had to send a heartbeat message
-    // to the client to keep the client-server connection alive
-    if (resultsFromServer == null || resultsFromServer.length == 0) {
-      // If this response was an empty heartbeat message, then we have not exhausted the region
-      // and thus there may be more partials server side that still need to be added to the partial
-      // list before we form the complete Result
-      if (!partialResults.isEmpty() && !heartbeatMessage) {
-        resultsToAddToCache.add(Result.createCompleteResult(partialResults));
-        clearPartialResults();
-      }
-
-      return resultsToAddToCache;
-    }
-
-    for(Result result : resultsFromServer) {
-      if (partialResultsRow != null && Bytes.compareTo(result.getRow(), partialResultsRow) != 0) {
-        // We have a new row, complete the previous row.
-        resultsToAddToCache.add(Result.createCompleteResult(partialResults));
-        clearPartialResults();
-      }
-      Result res = regroupResults(result);
-      if (res != null) {
-        resultsToAddToCache.add(res);
-      }
-      if (!result.mayHaveMoreCellsInRow()) {
-        // We are done for this row
-        if (partialResultsCellSizes > 0) {
-          resultsToAddToCache.add(Result.createCompleteResult(partialResults));
-        }
-        clearPartialResults();
-      }
-    }
-
-
-    return resultsToAddToCache;
-  }
-
-  /**
-   * Add new result to the partial list and return a batched Result if caching size exceed
-   * batching limit.
-   * If user setBatch(5) and rpc returns 3+5+5+5+3 cells, we should return 5+5+5+5+1 to user.
-   * setBatch doesn't mean setAllowPartialResult(true)
-   * @param result The result that we want to add to our list of partial Results
-   * @return the result if we have batch limit and there is one Result can be returned to user, or
-   *         null if we have not.
-   * @throws IOException
-   */
-  private Result regroupResults(final Result result) throws IOException {
-    partialResultsRow = result.getRow();
-    partialResults.add(result);
-    partialResultsCellSizes += result.size();
-    if (scan.getBatch() > 0 && partialResultsCellSizes >= scan.getBatch()) {
-      Cell[] cells = new Cell[scan.getBatch()];
-      int count = 0;
-      boolean stale = false;
-      while (count < scan.getBatch()) {
-        Result res = partialResults.poll();
-        stale = stale || res.isStale();
-        if (res.size() + count <= scan.getBatch()) {
-          System.arraycopy(res.rawCells(), 0, cells, count, res.size());
-          count += res.size();
-        } else {
-          int len = scan.getBatch() - count;
-          System.arraycopy(res.rawCells(), 0, cells, count, len);
-          Cell[] remainingCells = new Cell[res.size() - len];
-          System.arraycopy(res.rawCells(), len, remainingCells, 0, res.size() - len);
-          Result remainingRes = Result.create(remainingCells, res.getExists(), res.isStale(),
-              res.mayHaveMoreCellsInRow());
-          partialResults.addFirst(remainingRes);
-          count = scan.getBatch();
-        }
-      }
-      partialResultsCellSizes -= scan.getBatch();
-      if (partialResultsCellSizes == 0) {
-        // We have nothing in partialResults, clear the flags to prevent returning empty Result
-        // when next result belongs to the next row.
-        clearPartialResults();
-      }
-      return Result.create(cells, null, stale,
-          partialResultsCellSizes > 0 || result.mayHaveMoreCellsInRow());
-    }
-    return null;
-  }
-
-  /**
-   * Convenience method for clearing the list of partials and resetting the partialResultsRow.
-   */
-  private void clearPartialResults() {
-    partialResults.clear();
-    partialResultsCellSizes = 0;
-    partialResultsRow = null;
-  }
-
-  /**
-   * Helper method for adding results between the indices [start, end) to the outputList
-   * @param outputList the list that results will be added to
-   * @param inputArray the array that results are taken from
-   * @param start beginning index (inclusive)
-   * @param end ending index (exclusive)
-   */
-  private void addResultsToList(List<Result> outputList, Result[] inputArray, int start, int end) {
-    if (inputArray == null || start < 0 || end > inputArray.length) return;
-
-    for (int i = start; i < end; i++) {
-      outputList.add(inputArray[i]);
-    }
-  }
-
   @Override
   public void close() {
     if (!scanMetricsPublished) writeScanMetrics();
@@ -749,57 +571,6 @@ public abstract class ClientScanner extends AbstractClientScanner {
     return false;
   }
 
-  protected void updateLastCellLoadedToCache(Result result) {
-    if (result.rawCells().length == 0) {
-      return;
-    }
-    this.lastCellLoadedToCache = result.rawCells()[result.rawCells().length - 1];
-  }
-
-  /**
-   * Compare two Cells considering reversed scanner. ReversedScanner only reverses rows, not
-   * columns.
-   */
-  private int compare(Cell a, Cell b) {
-    CellComparator comparator = currentRegion != null && currentRegion.isMetaRegion()
-        ? CellComparator.META_COMPARATOR : CellComparator.COMPARATOR;
-    int r = comparator.compareRows(a, b);
-    if (r != 0) {
-      return this.scan.isReversed() ? -r : r;
-    }
-    return CellComparator.compareWithoutRow(a, b);
-  }
-
-  private Result filterLoadedCell(Result result) {
-    // we only filter result when last result is partial
-    // so lastCellLoadedToCache and result should have same row key.
-    // However, if 1) read some cells; 1.1) delete this row at the same time 2) move region;
-    // 3) read more cell. lastCellLoadedToCache and result will be not at same row.
-    if (lastCellLoadedToCache == null || result.rawCells().length == 0) {
-      return result;
-    }
-    if (compare(this.lastCellLoadedToCache, result.rawCells()[0]) < 0) {
-      // The first cell of this result is larger than the last cell of loadcache.
-      // If user do not allow partial result, it must be true.
-      return result;
-    }
-    if (compare(this.lastCellLoadedToCache, result.rawCells()[result.rawCells().length - 1]) >= 0) {
-      // The last cell of this result is smaller than the last cell of loadcache, skip all.
-      return null;
-    }
-
-    // The first one must not in filtered result, we start at the second.
-    int index = 1;
-    while (index < result.rawCells().length) {
-      if (compare(this.lastCellLoadedToCache, result.rawCells()[index]) < 0) {
-        break;
-      }
-      index++;
-    }
-    Cell[] list = Arrays.copyOfRange(result.rawCells(), index, result.rawCells().length);
-    return Result.create(list, result.getExists(), result.isStale(), result.mayHaveMoreCellsInRow());
-  }
-
   protected void initCache() {
     initSyncCache();
   }

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionUtils.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionUtils.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionUtils.java
index 2b75836..3e7cd00 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionUtils.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionUtils.java
@@ -316,6 +316,10 @@ public final class ConnectionUtils {
   }
 
   static Result filterCells(Result result, Cell keepCellsAfter) {
+    if (keepCellsAfter == null) {
+      // do not need to filter
+      return result;
+    }
     // not the same row
     if (!CellUtil.matchingRow(keepCellsAfter, result.getRow(), 0, result.getRow().length)) {
       return result;
@@ -410,4 +414,14 @@ public final class ConnectionUtils {
   public static int numberOfIndividualRows(List<Result> results) {
     return (int) results.stream().filter(r -> !r.mayHaveMoreCellsInRow()).count();
   }
+
+  public static ScanResultCache createScanResultCache(Scan scan) {
+    if (scan.getAllowPartialResults()) {
+      return new AllowPartialScanResultCache();
+    } else if (scan.getBatch() > 0) {
+      return new BatchScanResultCache(scan.getBatch());
+    } else {
+      return new CompleteScanResultCache();
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
index 4752d70..f8682ec 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Result.java
@@ -24,7 +24,9 @@ import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
@@ -145,11 +147,11 @@ public class Result implements CellScannable, CellScanner {
   }
 
   public static Result create(List<Cell> cells, Boolean exists, boolean stale,
-      boolean hasMoreCellsInRow) {
+      boolean mayHaveMoreCellsInRow) {
     if (exists != null){
-      return new Result(null, exists, stale, hasMoreCellsInRow);
+      return new Result(null, exists, stale, mayHaveMoreCellsInRow);
     }
-    return new Result(cells.toArray(new Cell[cells.size()]), null, stale, hasMoreCellsInRow);
+    return new Result(cells.toArray(new Cell[cells.size()]), null, stale, mayHaveMoreCellsInRow);
   }
 
   /**
@@ -792,44 +794,42 @@ public class Result implements CellScannable, CellScanner {
    * @throws IOException A complete result cannot be formed because the results in the partial list
    *           come from different rows
    */
-  public static Result createCompleteResult(List<Result> partialResults)
+  public static Result createCompleteResult(Iterable<Result> partialResults)
       throws IOException {
+    if (partialResults == null) {
+      return Result.create(Collections.emptyList(), null, false);
+    }
     List<Cell> cells = new ArrayList<>();
     boolean stale = false;
     byte[] prevRow = null;
     byte[] currentRow = null;
-
-    if (partialResults != null && !partialResults.isEmpty()) {
-      for (int i = 0; i < partialResults.size(); i++) {
-        Result r = partialResults.get(i);
-        currentRow = r.getRow();
-        if (prevRow != null && !Bytes.equals(prevRow, currentRow)) {
-          throw new IOException(
-              "Cannot form complete result. Rows of partial results do not match." +
-                  " Partial Results: " + partialResults);
-        }
-
-        // Ensure that all Results except the last one are marked as partials. The last result
-        // may not be marked as a partial because Results are only marked as partials when
-        // the scan on the server side must be stopped due to reaching the maxResultSize.
-        // Visualizing it makes it easier to understand:
-        // maxResultSize: 2 cells
-        // (-x-) represents cell number x in a row
-        // Example: row1: -1- -2- -3- -4- -5- (5 cells total)
-        // How row1 will be returned by the server as partial Results:
-        // Result1: -1- -2- (2 cells, size limit reached, mark as partial)
-        // Result2: -3- -4- (2 cells, size limit reached, mark as partial)
-        // Result3: -5- (1 cell, size limit NOT reached, NOT marked as partial)
-        if (i != (partialResults.size() - 1) && !r.mayHaveMoreCellsInRow()) {
-          throw new IOException(
-              "Cannot form complete result. Result is missing partial flag. " +
-                  "Partial Results: " + partialResults);
-        }
-        prevRow = currentRow;
-        stale = stale || r.isStale();
-        for (Cell c : r.rawCells()) {
-          cells.add(c);
-        }
+    for (Iterator<Result> iter = partialResults.iterator(); iter.hasNext();) {
+      Result r = iter.next();
+      currentRow = r.getRow();
+      if (prevRow != null && !Bytes.equals(prevRow, currentRow)) {
+        throw new IOException(
+            "Cannot form complete result. Rows of partial results do not match." +
+                " Partial Results: " + partialResults);
+      }
+      // Ensure that all Results except the last one are marked as partials. The last result
+      // may not be marked as a partial because Results are only marked as partials when
+      // the scan on the server side must be stopped due to reaching the maxResultSize.
+      // Visualizing it makes it easier to understand:
+      // maxResultSize: 2 cells
+      // (-x-) represents cell number x in a row
+      // Example: row1: -1- -2- -3- -4- -5- (5 cells total)
+      // How row1 will be returned by the server as partial Results:
+      // Result1: -1- -2- (2 cells, size limit reached, mark as partial)
+      // Result2: -3- -4- (2 cells, size limit reached, mark as partial)
+      // Result3: -5- (1 cell, size limit NOT reached, NOT marked as partial)
+      if (iter.hasNext() && !r.mayHaveMoreCellsInRow()) {
+        throw new IOException("Cannot form complete result. Result is missing partial flag. " +
+            "Partial Results: " + partialResults);
+      }
+      prevRow = currentRow;
+      stale = stale || r.isStale();
+      for (Cell c : r.rawCells()) {
+        cells.add(c);
       }
     }
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAllowPartialScanResultCache.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAllowPartialScanResultCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAllowPartialScanResultCache.java
index fc5ba14..3fe43a5 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAllowPartialScanResultCache.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAllowPartialScanResultCache.java
@@ -17,14 +17,14 @@
  */
 package org.apache.hadoop.hbase.client;
 
-import static org.junit.Assert.*;
+import static org.apache.hadoop.hbase.client.TestBatchScanResultCache.createCells;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.stream.IntStream;
 
 import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.testclassification.ClientTests;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -51,10 +51,6 @@ public class TestAllowPartialScanResultCache {
     resultCache = null;
   }
 
-  private static Cell createCell(int key, int cq) {
-    return new KeyValue(Bytes.toBytes(key), CF, Bytes.toBytes("cq" + cq), Bytes.toBytes(key));
-  }
-
   @Test
   public void test() throws IOException {
     assertSame(ScanResultCache.EMPTY_RESULT_ARRAY,
@@ -62,31 +58,34 @@ public class TestAllowPartialScanResultCache {
     assertSame(ScanResultCache.EMPTY_RESULT_ARRAY,
       resultCache.addAndGet(ScanResultCache.EMPTY_RESULT_ARRAY, true));
 
-    Cell[] cells1 = IntStream.range(0, 10).mapToObj(i -> createCell(1, i)).toArray(Cell[]::new);
-    Cell[] cells2 = IntStream.range(0, 10).mapToObj(i -> createCell(2, i)).toArray(Cell[]::new);
+    Cell[] cells1 = createCells(CF, 1, 10);
+    Cell[] cells2 = createCells(CF, 2, 10);
 
     Result[] results1 = resultCache.addAndGet(
       new Result[] { Result.create(Arrays.copyOf(cells1, 5), null, false, true) }, false);
     assertEquals(1, results1.length);
     assertEquals(1, Bytes.toInt(results1[0].getRow()));
     assertEquals(5, results1[0].rawCells().length);
-    IntStream.range(0, 5).forEach(
-      i -> assertEquals(1, Bytes.toInt(results1[0].getValue(CF, Bytes.toBytes("cq" + i)))));
+    for (int i = 0; i < 5; i++) {
+      assertEquals(1, Bytes.toInt(results1[0].getValue(CF, Bytes.toBytes("cq" + i))));
+    }
 
     Result[] results2 = resultCache.addAndGet(
       new Result[] { Result.create(Arrays.copyOfRange(cells1, 1, 10), null, false, true) }, false);
     assertEquals(1, results2.length);
     assertEquals(1, Bytes.toInt(results2[0].getRow()));
     assertEquals(5, results2[0].rawCells().length);
-    IntStream.range(5, 10).forEach(
-      i -> assertEquals(1, Bytes.toInt(results2[0].getValue(CF, Bytes.toBytes("cq" + i)))));
+    for (int i = 5; i < 10; i++) {
+      assertEquals(1, Bytes.toInt(results2[0].getValue(CF, Bytes.toBytes("cq" + i))));
+    }
 
-    Result[] results3 = resultCache
-        .addAndGet(new Result[] { Result.create(cells1), Result.create(cells2) }, false);
+    Result[] results3 =
+        resultCache.addAndGet(new Result[] { Result.create(cells1), Result.create(cells2) }, false);
     assertEquals(1, results3.length);
     assertEquals(2, Bytes.toInt(results3[0].getRow()));
     assertEquals(10, results3[0].rawCells().length);
-    IntStream.range(0, 10).forEach(
-      i -> assertEquals(2, Bytes.toInt(results3[0].getValue(CF, Bytes.toBytes("cq" + i)))));
+    for (int i = 0; i < 10; i++) {
+      assertEquals(2, Bytes.toInt(results3[0].getValue(CF, Bytes.toBytes("cq" + i))));
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBatchScanResultCache.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBatchScanResultCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBatchScanResultCache.java
new file mode 100644
index 0000000..31a4594
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBatchScanResultCache.java
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category({ SmallTests.class, ClientTests.class })
+public class TestBatchScanResultCache {
+
+  private static byte[] CF = Bytes.toBytes("cf");
+
+  private BatchScanResultCache resultCache;
+
+  @Before
+  public void setUp() {
+    resultCache = new BatchScanResultCache(4);
+  }
+
+  @After
+  public void tearDown() {
+    resultCache.clear();
+    resultCache = null;
+  }
+
+  static Cell createCell(byte[] cf, int key, int cq) {
+    return new KeyValue(Bytes.toBytes(key), cf, Bytes.toBytes("cq" + cq), Bytes.toBytes(key));
+  }
+
+  static Cell[] createCells(byte[] cf, int key, int numCqs) {
+    Cell[] cells = new Cell[numCqs];
+    for (int i = 0; i < numCqs; i++) {
+      cells[i] = createCell(cf, key, i);
+    }
+    return cells;
+  }
+
+  private void assertResultEquals(Result result, int key, int start, int to) {
+    assertEquals(to - start, result.size());
+    for (int i = start; i < to; i++) {
+      assertEquals(key, Bytes.toInt(result.getValue(CF, Bytes.toBytes("cq" + i))));
+    }
+    assertEquals(to - start == 4, result.mayHaveMoreCellsInRow());
+  }
+
+  @Test
+  public void test() throws IOException {
+    assertSame(ScanResultCache.EMPTY_RESULT_ARRAY,
+      resultCache.addAndGet(ScanResultCache.EMPTY_RESULT_ARRAY, false));
+    assertSame(ScanResultCache.EMPTY_RESULT_ARRAY,
+      resultCache.addAndGet(ScanResultCache.EMPTY_RESULT_ARRAY, true));
+
+    Cell[] cells1 = createCells(CF, 1, 10);
+    Cell[] cells2 = createCells(CF, 2, 10);
+    Cell[] cells3 = createCells(CF, 3, 10);
+    assertEquals(0, resultCache.addAndGet(
+      new Result[] { Result.create(Arrays.copyOf(cells1, 3), null, false, true) }, false).length);
+    Result[] results = resultCache.addAndGet(
+      new Result[] { Result.create(Arrays.copyOfRange(cells1, 3, 7), null, false, true),
+          Result.create(Arrays.copyOfRange(cells1, 7, 10), null, false, true) },
+      false);
+    assertEquals(2, results.length);
+    assertResultEquals(results[0], 1, 0, 4);
+    assertResultEquals(results[1], 1, 4, 8);
+    results = resultCache.addAndGet(ScanResultCache.EMPTY_RESULT_ARRAY, false);
+    assertEquals(1, results.length);
+    assertResultEquals(results[0], 1, 8, 10);
+
+    results = resultCache.addAndGet(
+      new Result[] { Result.create(Arrays.copyOfRange(cells2, 0, 4), null, false, true),
+          Result.create(Arrays.copyOfRange(cells2, 4, 8), null, false, true),
+          Result.create(Arrays.copyOfRange(cells2, 8, 10), null, false, true),
+          Result.create(Arrays.copyOfRange(cells3, 0, 4), null, false, true),
+          Result.create(Arrays.copyOfRange(cells3, 4, 8), null, false, true),
+          Result.create(Arrays.copyOfRange(cells3, 8, 10), null, false, false) },
+      false);
+    assertEquals(6, results.length);
+    assertResultEquals(results[0], 2, 0, 4);
+    assertResultEquals(results[1], 2, 4, 8);
+    assertResultEquals(results[2], 2, 8, 10);
+    assertResultEquals(results[3], 3, 0, 4);
+    assertResultEquals(results[4], 3, 4, 8);
+    assertResultEquals(results[5], 3, 8, 10);
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestCompleteResultScanResultCache.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestCompleteResultScanResultCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestCompleteResultScanResultCache.java
index a340e9f..8759593 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestCompleteResultScanResultCache.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestCompleteResultScanResultCache.java
@@ -22,7 +22,6 @@ import static org.junit.Assert.assertSame;
 
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.stream.IntStream;
 
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.KeyValue;
@@ -70,9 +69,9 @@ public class TestCompleteResultScanResultCache {
       resultCache.addAndGet(ScanResultCache.EMPTY_RESULT_ARRAY, true));
     int count = 10;
     Result[] results = new Result[count];
-    IntStream.range(0, count).forEach(i -> {
+    for (int i = 0; i < count; i++) {
       results[i] = Result.create(Arrays.asList(createCell(i, CQ1)));
-    });
+    }
     assertSame(results, resultCache.addAndGet(results, false));
   }
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/1849e8a5/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTablePartialScan.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTablePartialScan.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTablePartialScan.java
new file mode 100644
index 0000000..2a32206
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestRawAsyncTablePartialScan.java
@@ -0,0 +1,119 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category({ MediumTests.class, ClientTests.class })
+public class TestRawAsyncTablePartialScan {
+
+  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+
+  private static TableName TABLE_NAME = TableName.valueOf("async");
+
+  private static byte[] FAMILY = Bytes.toBytes("cf");
+
+  private static byte[][] CQS =
+      new byte[][] { Bytes.toBytes("cq1"), Bytes.toBytes("cq2"), Bytes.toBytes("cq3") };
+
+  private static int COUNT = 100;
+
+  private static AsyncConnection CONN;
+
+  private static RawAsyncTable TABLE;
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    TEST_UTIL.startMiniCluster(1);
+    TEST_UTIL.createTable(TABLE_NAME, FAMILY);
+    TEST_UTIL.waitTableAvailable(TABLE_NAME);
+    CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
+    TABLE = CONN.getRawTable(TABLE_NAME);
+    TABLE
+        .putAll(IntStream.range(0, COUNT)
+            .mapToObj(i -> new Put(Bytes.toBytes(String.format("%02d", i)))
+                .addColumn(FAMILY, CQS[0], Bytes.toBytes(i))
+                .addColumn(FAMILY, CQS[1], Bytes.toBytes(2 * i))
+                .addColumn(FAMILY, CQS[2], Bytes.toBytes(3 * i)))
+            .collect(Collectors.toList()))
+        .get();
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    CONN.close();
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Test
+  public void testBatchDoNotAllowPartial() throws InterruptedException, ExecutionException {
+    // we set batch to 2 and max result size to 1, then server will only returns one result per call
+    // but we should get 2 + 1 for every row.
+    List<Result> results = TABLE.scanAll(new Scan().setBatch(2).setMaxResultSize(1)).get();
+    assertEquals(2 * COUNT, results.size());
+    for (int i = 0; i < COUNT; i++) {
+      Result firstTwo = results.get(2 * i);
+      assertEquals(String.format("%02d", i), Bytes.toString(firstTwo.getRow()));
+      assertEquals(2, firstTwo.size());
+      assertEquals(i, Bytes.toInt(firstTwo.getValue(FAMILY, CQS[0])));
+      assertEquals(2 * i, Bytes.toInt(firstTwo.getValue(FAMILY, CQS[1])));
+
+      Result secondOne = results.get(2 * i + 1);
+      assertEquals(String.format("%02d", i), Bytes.toString(secondOne.getRow()));
+      assertEquals(1, secondOne.size());
+      assertEquals(3 * i, Bytes.toInt(secondOne.getValue(FAMILY, CQS[2])));
+    }
+  }
+
+  @Test
+  public void testReversedBatchDoNotAllowPartial() throws InterruptedException, ExecutionException {
+    // we set batch to 2 and max result size to 1, then server will only returns one result per call
+    // but we should get 2 + 1 for every row.
+    List<Result> results =
+        TABLE.scanAll(new Scan().setBatch(2).setMaxResultSize(1).setReversed(true)).get();
+    assertEquals(2 * COUNT, results.size());
+    for (int i = 0; i < COUNT; i++) {
+      int row = COUNT - i - 1;
+      Result firstTwo = results.get(2 * i);
+      assertEquals(String.format("%02d", row), Bytes.toString(firstTwo.getRow()));
+      assertEquals(2, firstTwo.size());
+      assertEquals(row, Bytes.toInt(firstTwo.getValue(FAMILY, CQS[0])));
+      assertEquals(2 * row, Bytes.toInt(firstTwo.getValue(FAMILY, CQS[1])));
+
+      Result secondOne = results.get(2 * i + 1);
+      assertEquals(String.format("%02d", row), Bytes.toString(secondOne.getRow()));
+      assertEquals(1, secondOne.size());
+      assertEquals(3 * row, Bytes.toInt(secondOne.getValue(FAMILY, CQS[2])));
+    }
+  }
+}


Mime
View raw message