1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.client;
20
21 import java.io.IOException;
22 import java.util.concurrent.ExecutorService;
23
24 import org.apache.commons.logging.Log;
25 import org.apache.commons.logging.LogFactory;
26 import org.apache.hadoop.hbase.classification.InterfaceAudience;
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.hbase.HConstants;
29 import org.apache.hadoop.hbase.TableName;
30 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
31 import org.apache.hadoop.hbase.util.Bytes;
32 import org.apache.hadoop.hbase.util.ExceptionUtil;
33
34
35
36
37 @InterfaceAudience.Private
38 public class ReversedClientScanner extends ClientScanner {
39 private static final Log LOG = LogFactory.getLog(ReversedClientScanner.class);
40
41
42
43
44
45
46
47
48
49
50
51
52 public ReversedClientScanner(Configuration conf, Scan scan,
53 TableName tableName, ClusterConnection connection,
54 RpcRetryingCallerFactory rpcFactory, RpcControllerFactory controllerFactory,
55 ExecutorService pool, int primaryOperationTimeout) throws IOException {
56 super(conf, scan, tableName, connection, rpcFactory, controllerFactory, pool,
57 primaryOperationTimeout);
58 }
59
60 @Override
61 protected boolean nextScanner(int nbRows, final boolean done)
62 throws IOException {
63
64 closeScanner();
65
66
67 byte[] localStartKey;
68 boolean locateTheClosestFrontRow = true;
69
70 if (this.currentRegion != null) {
71 byte[] startKey = this.currentRegion.getStartKey();
72 if (startKey == null
73 || Bytes.equals(startKey, HConstants.EMPTY_BYTE_ARRAY)
74 || checkScanStopRow(startKey) || done) {
75 close();
76 if (LOG.isDebugEnabled()) {
77 LOG.debug("Finished " + this.currentRegion);
78 }
79 return false;
80 }
81 localStartKey = startKey;
82 if (LOG.isDebugEnabled()) {
83 LOG.debug("Finished " + this.currentRegion);
84 }
85 } else {
86 localStartKey = this.scan.getStartRow();
87 if (!Bytes.equals(localStartKey, HConstants.EMPTY_BYTE_ARRAY)) {
88 locateTheClosestFrontRow = false;
89 }
90 }
91
92 if (LOG.isDebugEnabled() && this.currentRegion != null) {
93
94 LOG.debug("Advancing internal scanner to startKey at '"
95 + Bytes.toStringBinary(localStartKey) + "'");
96 }
97 try {
98
99
100
101
102
103
104
105 byte[] locateStartRow = locateTheClosestFrontRow ? createClosestRowBefore(localStartKey)
106 : null;
107 callable = getScannerCallable(localStartKey, nbRows, locateStartRow);
108
109
110
111
112 this.caller.callWithoutRetries(callable, scannerTimeout);
113 this.currentRegion = callable.getHRegionInfo();
114 if (this.scanMetrics != null) {
115 this.scanMetrics.countOfRegions.incrementAndGet();
116 }
117 } catch (IOException e) {
118 ExceptionUtil.rethrowIfInterrupt(e);
119 close();
120 throw e;
121 }
122 return true;
123 }
124
125 protected ScannerCallableWithReplicas getScannerCallable(byte[] localStartKey,
126 int nbRows, byte[] locateStartRow) {
127 scan.setStartRow(localStartKey);
128 ScannerCallable s =
129 new ReversedScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
130 locateStartRow, this.rpcControllerFactory);
131 s.setCaching(nbRows);
132 ScannerCallableWithReplicas sr =
133 new ScannerCallableWithReplicas(getTable(), getConnection(), s, pool,
134 primaryOperationTimeout, scan, getRetries(), getScannerTimeout(), caching, getConf(),
135 caller);
136 return sr;
137 }
138
139 @Override
140
141 protected boolean checkScanStopRow(final byte[] startKey) {
142 if (this.scan.getStopRow().length > 0) {
143
144 byte[] stopRow = scan.getStopRow();
145 int cmp = Bytes.compareTo(stopRow, 0, stopRow.length, startKey, 0,
146 startKey.length);
147 if (cmp >= 0) {
148
149
150 return true;
151 }
152 }
153 return false;
154 }
155 }