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.classification.InterfaceAudience;
26  import org.apache.hadoop.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.ipc.RpcControllerFactory;
33  import org.apache.hadoop.hbase.util.Bytes;
34  
35  import java.io.IOException;
36  import java.util.concurrent.ExecutorService;
37  
38  /**
39   * Client scanner for small reversed scan. Generally, only one RPC is called to fetch the
40   * scan results, unless the results cross multiple regions or the row count of
41   * results exceed the caching.
42   * <p/>
43   * For small scan, it will get better performance than {@link ReversedClientScanner}
44   */
45  @InterfaceAudience.Private
46  public class ClientSmallReversedScanner extends ReversedClientScanner {
47    private static final Log LOG = LogFactory.getLog(ClientSmallReversedScanner.class);
48    private ScannerCallableWithReplicas smallScanCallable = null;
49    private byte[] skipRowOfFirstResult = null;
50  
51    /**
52     * Create a new ReversibleClientScanner for the specified table Note that the
53     * passed {@link Scan}'s start row maybe changed changed.
54     *
55     * @param conf The {@link Configuration} to use.
56     * @param scan {@link Scan} to use in this scanner
57     * @param tableName The table that we wish to rangeGet
58     * @param connection Connection identifying the cluster
59     * @param rpcFactory
60     * @throws IOException
61     */
62    public ClientSmallReversedScanner(final Configuration conf, final Scan scan,
63        final TableName tableName, ClusterConnection connection,
64        RpcRetryingCallerFactory rpcFactory, RpcControllerFactory controllerFactory,
65        ExecutorService pool, int primaryOperationTimeout) throws IOException {
66      super(conf, scan, tableName, connection, rpcFactory, controllerFactory, pool, primaryOperationTimeout);
67    }
68  
69    /**
70     * Gets a scanner for following scan. Move to next region or continue from the
71     * last result or start from the start row.
72     *
73     * @param nbRows
74     * @param done              true if Server-side says we're done scanning.
75     * @param currentRegionDone true if scan is over on current region
76     * @return true if has next scanner
77     * @throws IOException
78     */
79    private boolean nextScanner(int nbRows, final boolean done,
80                                boolean currentRegionDone) throws IOException {
81      // Where to start the next getter
82      byte[] localStartKey;
83      int cacheNum = nbRows;
84      skipRowOfFirstResult = null;
85      // if we're at end of table, close and return false to stop iterating
86      if (this.currentRegion != null && currentRegionDone) {
87        byte[] startKey = this.currentRegion.getStartKey();
88        if (startKey == null
89            || Bytes.equals(startKey, HConstants.EMPTY_BYTE_ARRAY)
90            || checkScanStopRow(startKey) || done) {
91          close();
92          if (LOG.isDebugEnabled()) {
93            LOG.debug("Finished with small scan at " + this.currentRegion);
94          }
95          return false;
96        }
97        // We take the row just under to get to the previous region.
98        localStartKey = createClosestRowBefore(startKey);
99        if (LOG.isDebugEnabled()) {
100         LOG.debug("Finished with region " + this.currentRegion);
101       }
102     } else if (this.lastResult != null) {
103       localStartKey = this.lastResult.getRow();
104       skipRowOfFirstResult = this.lastResult.getRow();
105       cacheNum++;
106     } else {
107       localStartKey = this.scan.getStartRow();
108     }
109 
110     if (LOG.isTraceEnabled()) {
111       LOG.trace("Advancing internal small scanner to startKey at '"
112           + Bytes.toStringBinary(localStartKey) + "'");
113     }
114 
115     smallScanCallable = ClientSmallScanner.getSmallScanCallable(
116       getConnection(), getTable(), scan, getScanMetrics(), localStartKey, cacheNum,
117       rpcControllerFactory, getPool(), getPrimaryOperationTimeout(),
118       getRetries(), getScannerTimeout(), getConf(), caller);
119 
120     if (this.scanMetrics != null && skipRowOfFirstResult == null) {
121       this.scanMetrics.countOfRegions.incrementAndGet();
122     }
123     return true;
124   }
125 
126   @Override
127   public Result next() throws IOException {
128     // If the scanner is closed and there's nothing left in the cache, next is a
129     // no-op.
130     if (cache.size() == 0 && this.closed) {
131       return null;
132     }
133     if (cache.size() == 0) {
134       Result[] values = null;
135       long remainingResultSize = maxScannerResultSize;
136       int countdown = this.caching;
137       boolean currentRegionDone = false;
138       // Values == null means server-side filter has determined we must STOP
139       while (remainingResultSize > 0 && countdown > 0
140           && nextScanner(countdown, values == null, currentRegionDone)) {
141         // Server returns a null values if scanning is to stop. Else,
142         // returns an empty array if scanning is to go on and we've just
143         // exhausted current region.
144         // callWithoutRetries is at this layer. Within the ScannerCallableWithReplicas,
145         // we do a callWithRetries
146         values = this.caller.callWithoutRetries(smallScanCallable, scannerTimeout);
147         this.currentRegion = smallScanCallable.getHRegionInfo();
148         long currentTime = System.currentTimeMillis();
149         if (this.scanMetrics != null) {
150           this.scanMetrics.sumOfMillisSecBetweenNexts.addAndGet(currentTime
151               - lastNext);
152         }
153         lastNext = currentTime;
154         if (values != null && values.length > 0) {
155           for (int i = 0; i < values.length; i++) {
156             Result rs = values[i];
157             if (i == 0 && this.skipRowOfFirstResult != null
158                 && Bytes.equals(skipRowOfFirstResult, rs.getRow())) {
159               // Skip the first result
160               continue;
161             }
162             cache.add(rs);
163             for (Cell kv : rs.rawCells()) {
164               remainingResultSize -= KeyValueUtil.ensureKeyValue(kv).heapSize();
165             }
166             countdown--;
167             this.lastResult = rs;
168           }
169         }
170         currentRegionDone = countdown > 0;
171       }
172     }
173 
174     if (cache.size() > 0) {
175       return cache.poll();
176     }
177     // if we exhausted this scanner before calling close, write out the scan
178     // metrics
179     writeScanMetrics();
180     return null;
181   }
182 
183 
184   @Override
185   protected void initializeScannerInConstruction() throws IOException {
186     // No need to initialize the scanner when constructing instance, do it when
187     // calling next(). Do nothing here.
188   }
189 
190   @Override
191   public void close() {
192     if (!scanMetricsPublished) writeScanMetrics();
193     closed = true;
194   }
195 
196 }