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