hbase-client/src/main/java/org/apache/hadoop/hbase/client/CompleteScanResultCache.java

   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.util.ArrayList;
  22 import java.util.Arrays;
  23 import java.util.List;
  24
  25 import org.apache.yetus.audience.InterfaceAudience;
  26 import org.apache.hadoop.hbase.util.Bytes;
  27
  28 /**
  29  * A scan result cache that only returns complete result.
  30  */
  31 @InterfaceAudience.Private
  32 class CompleteScanResultCache implements ScanResultCache {
  33
  34   private int numberOfCompleteRows;
  35
  36   private final List<Result> partialResults = new ArrayList<>();
  37
  38   private Result combine() throws IOException {
  39     Result result = Result.createCompleteResult(partialResults);
  40     partialResults.clear();
  41     return result;
  42   }
  43
  44   private Result[] prependCombined(Result[] results, int length) throws IOException {
  45     if (length == 0) {
  46       return new Result[] { combine() };
  47     }
  48     // the last part of a partial result may not be marked as partial so here we need to check if
  49     // there is a row change.
  50     int start;
  51     if (Bytes.equals(partialResults.get(0).getRow(), results[0].getRow())) {
  52       partialResults.add(results[0]);
  53       start = 1;
  54       length--;
  55     } else {
  56       start = 0;
  57     }
  58     Result[] prependResults = new Result[length + 1];
  59     prependResults[0] = combine();
  60     System.arraycopy(results, start, prependResults, 1, length);
  61     return prependResults;
  62   }
  63
  64   private Result[] updateNumberOfCompleteResultsAndReturn(Result... results) {
  65     numberOfCompleteRows += results.length;
  66     return results;
  67   }
  68
  69   @Override
  70   public Result[] addAndGet(Result[] results, boolean isHeartbeatMessage) throws IOException {
  71     // If no results were returned it indicates that either we have the all the partial results
  72     // necessary to construct the complete result or the server had to send a heartbeat message
  73     // to the client to keep the client-server connection alive
  74     if (results.length == 0) {
  75       // If this response was an empty heartbeat message, then we have not exhausted the region
  76       // and thus there may be more partials server side that still need to be added to the partial
  77       // list before we form the complete Result
  78       if (!partialResults.isEmpty() && !isHeartbeatMessage) {
  79         return updateNumberOfCompleteResultsAndReturn(combine());
  80       }
  81       return EMPTY_RESULT_ARRAY;
  82     }
  83     // In every RPC response there should be at most a single partial result. Furthermore, if
  84     // there is a partial result, it is guaranteed to be in the last position of the array.
  85     Result last = results[results.length - 1];
  86     if (last.mayHaveMoreCellsInRow()) {
  87       if (partialResults.isEmpty()) {
  88         partialResults.add(last);
  89         return updateNumberOfCompleteResultsAndReturn(Arrays.copyOf(results, results.length - 1));
  90       }
  91       // We have only one result and it is partial
  92       if (results.length == 1) {
  93         // check if there is a row change
  94         if (Bytes.equals(partialResults.get(0).getRow(), last.getRow())) {
  95           partialResults.add(last);
  96           return EMPTY_RESULT_ARRAY;
  97         }
  98         Result completeResult = combine();
  99         partialResults.add(last);
 100         return updateNumberOfCompleteResultsAndReturn(completeResult);
 101       }
 102       // We have some complete results
 103       Result[] resultsToReturn = prependCombined(results, results.length - 1);
 104       partialResults.add(last);
 105       return updateNumberOfCompleteResultsAndReturn(resultsToReturn);
 106     }
 107     if (!partialResults.isEmpty()) {
 108       return updateNumberOfCompleteResultsAndReturn(prependCombined(results, results.length));
 109     }
 110     return updateNumberOfCompleteResultsAndReturn(results);
 111   }
 112
 113   @Override
 114   public void clear() {
 115     partialResults.clear();
 116   }
 117
 118   @Override
 119   public int numberOfCompleteRows() {
 120     return numberOfCompleteRows;
 121   }
 122 }