View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one or more
5    * contributor license agreements. See the NOTICE file distributed with this
6    * work for additional information regarding copyright ownership. The ASF
7    * licenses this file to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance with the License.
9    * You may obtain a copy of the License at
10   *
11   * http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16   * License for the specific language governing permissions and limitations
17   * under the License.
18   */
19  
20  package org.apache.hadoop.hbase.client;
21  
22  
23  import org.apache.commons.logging.Log;
24  import org.apache.commons.logging.LogFactory;
25  import org.apache.hadoop.hbase.classification.InterfaceAudience;
26  import org.apache.hadoop.hbase.classification.InterfaceStability;
27  import org.apache.hadoop.conf.Configuration;
28  import org.apache.hadoop.hbase.Cell;
29  import org.apache.hadoop.hbase.HConstants;
30  import org.apache.hadoop.hbase.KeyValueUtil;
31  import org.apache.hadoop.hbase.TableName;
32  import org.apache.hadoop.hbase.client.ClientSmallScanner.SmallScannerCallableFactory;
33  import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
34  import org.apache.hadoop.hbase.util.Bytes;
35  
36  import com.google.common.annotations.VisibleForTesting;
37  
38  import java.io.IOException;
39  
40  /**
41   * Client scanner for small reversed scan. Generally, only one RPC is called to fetch the
42   * scan results, unless the results cross multiple regions or the row count of
43   * results exceed the caching.
44   * <p/>
45   * For small scan, it will get better performance than {@link ReversedClientScanner}
46   */
47  @InterfaceAudience.Public
48  @InterfaceStability.Evolving
49  public class ClientSmallReversedScanner extends ReversedClientScanner {
50    private static final Log LOG = LogFactory.getLog(ClientSmallReversedScanner.class);
51    private RegionServerCallable<Result[]> smallScanCallable = null;
52    private SmallScannerCallableFactory callableFactory;
53  
54    /**
55     * Create a new ReversibleClientScanner for the specified table Note that the
56     * passed {@link org.apache.hadoop.hbase.client.Scan}'s start row maybe changed.
57     *
58     * @param conf       The {@link org.apache.hadoop.conf.Configuration} to use.
59     * @param scan       {@link org.apache.hadoop.hbase.client.Scan} to use in this scanner
60     * @param tableName  The table that we wish to scan
61     * @param connection Connection identifying the cluster
62     * @throws java.io.IOException
63     */
64    public ClientSmallReversedScanner(Configuration conf, Scan scan, TableName tableName,
65                                      HConnection connection) throws IOException {
66      this(conf, scan, tableName, connection, new SmallScannerCallableFactory());
67    }
68  
69    /**
70     * Create a new ReversibleClientScanner for the specified table. Take note that the passed
71     * {@link Scan}'s start row may be changed.
72     *
73     * @param conf
74     *          The {@link Configuration} to use.
75     * @param scan
76     *          {@link Scan} to use in this scanner
77     * @param tableName
78     *          The table that we wish to rangeGet
79     * @param connection
80     *          Connection identifying the cluster
81     * @param callableFactory
82     *          Factory used to create the callable for scans
83     * @throws IOException
84     *           If the remote call fails
85     */
86    @VisibleForTesting
87    ClientSmallReversedScanner(Configuration conf, Scan scan, TableName tableName,
88        HConnection connection, SmallScannerCallableFactory callableFactory) throws IOException {
89      super(conf, scan, tableName, connection);
90      this.callableFactory = callableFactory;
91    }
92  
93    /**
94     * Gets a scanner for following scan. Move to next region or continue from the last result or
95     * start from the start row.
96     *
97     * @param nbRows
98     * @param done
99     *          true if Server-side says we're done scanning.
100    * @param currentRegionDone
101    *          true if scan is over on current region
102    * @return true if has next scanner
103    * @throws IOException
104    */
105   private boolean nextScanner(int nbRows, final boolean done,
106                               boolean currentRegionDone) throws IOException {
107     // Where to start the next getter
108     byte[] localStartKey;
109     int cacheNum = nbRows;
110     boolean regionChanged = true;
111     // if we're at end of table, close and return false to stop iterating
112     if (this.currentRegion != null && currentRegionDone) {
113       byte[] startKey = this.currentRegion.getStartKey();
114       if (startKey == null
115           || Bytes.equals(startKey, HConstants.EMPTY_BYTE_ARRAY)
116           || checkScanStopRow(startKey) || done) {
117         close();
118         if (LOG.isDebugEnabled()) {
119           LOG.debug("Finished with small scan at " + this.currentRegion);
120         }
121         return false;
122       }
123       // We take the row just under to get to the previous region.
124       localStartKey = createClosestRowBefore(startKey);
125       if (LOG.isDebugEnabled()) {
126         LOG.debug("Finished with region " + this.currentRegion);
127       }
128     } else if (this.lastResult != null) {
129       regionChanged = false;
130       localStartKey = createClosestRowBefore(lastResult.getRow());
131     } else {
132       localStartKey = this.scan.getStartRow();
133     }
134 
135     if (LOG.isTraceEnabled()) {
136       LOG.trace("Advancing internal small scanner to startKey at '"
137           + Bytes.toStringBinary(localStartKey) + "'");
138     }
139 
140     smallScanCallable = callableFactory.getCallable(
141         scan, getConnection(), getTable(), localStartKey, cacheNum, this.rpcControllerFactory);
142 
143     if (this.scanMetrics != null && regionChanged) {
144       this.scanMetrics.countOfRegions.incrementAndGet();
145     }
146     return true;
147   }
148 
149   @Override
150   public Result next() throws IOException {
151     // If the scanner is closed and there's nothing left in the cache, next is a
152     // no-op.
153     if (cache.size() == 0 && this.closed) {
154       return null;
155     }
156     if (cache.size() == 0) {
157       loadCache();
158     }
159 
160     if (cache.size() > 0) {
161       return cache.poll();
162     }
163     // if we exhausted this scanner before calling close, write out the scan
164     // metrics
165     writeScanMetrics();
166     return null;
167   }
168 
169   @Override
170   protected void loadCache() throws IOException {
171     Result[] values = null;
172     long remainingResultSize = maxScannerResultSize;
173     int countdown = this.caching;
174     boolean currentRegionDone = false;
175     // Values == null means server-side filter has determined we must STOP
176     while (remainingResultSize > 0 && countdown > 0
177         && nextScanner(countdown, values == null, currentRegionDone)) {
178       // Server returns a null values if scanning is to stop. Else,
179       // returns an empty array if scanning is to go on and we've just
180       // exhausted current region.
181       values = this.caller.callWithRetries(smallScanCallable, scannerTimeout);
182       this.currentRegion = smallScanCallable.getHRegionInfo();
183       long currentTime = System.currentTimeMillis();
184       if (this.scanMetrics != null) {
185         this.scanMetrics.sumOfMillisSecBetweenNexts.addAndGet(currentTime
186             - lastNext);
187       }
188       lastNext = currentTime;
189       if (values != null && values.length > 0) {
190         for (int i = 0; i < values.length; i++) {
191           Result rs = values[i];
192           cache.add(rs);
193           for (Cell kv : rs.rawCells()) {
194             remainingResultSize -= KeyValueUtil.ensureKeyValue(kv).heapSize();
195           }
196           countdown--;
197           this.lastResult = rs;
198         }
199       }
200       if (smallScanCallable.hasMoreResultsContext()) {
201         currentRegionDone = !smallScanCallable.getServerHasMoreResults();
202       } else {
203         currentRegionDone = countdown > 0;
204       }
205     }
206   }
207 
208   @Override
209   protected void initializeScannerInConstruction() throws IOException {
210     // No need to initialize the scanner when constructing instance, do it when
211     // calling next(). Do nothing here.
212   }
213 
214   @Override
215   public void close() {
216     if (!scanMetricsPublished) writeScanMetrics();
217     closed = true;
218   }
219 
220   @VisibleForTesting
221   protected void setScannerCallableFactory(SmallScannerCallableFactory callableFactory) {
222     this.callableFactory = callableFactory;
223   }
224 
225   @VisibleForTesting
226   protected void setRpcRetryingCaller(RpcRetryingCaller<Result []> caller) {
227     this.caller = caller;
228   }
229 
230   @VisibleForTesting
231   protected void setRpcControllerFactory(RpcControllerFactory rpcControllerFactory) {
232     this.rpcControllerFactory = rpcControllerFactory;
233   }
234 }