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.ArrayList;
23 import java.util.List;
24
25 import org.apache.hadoop.hbase.classification.InterfaceAudience;
26 import org.apache.hadoop.hbase.classification.InterfaceStability;
27 import org.apache.hadoop.hbase.DoNotRetryIOException;
28 import org.apache.hadoop.hbase.HConstants;
29 import org.apache.hadoop.hbase.HRegionLocation;
30 import org.apache.hadoop.hbase.TableName;
31 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
32 import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
33 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
34 import org.apache.hadoop.hbase.util.Bytes;
35
36
37
38
39 @InterfaceAudience.Public
40 @InterfaceStability.Evolving
41 public class ReversedScannerCallable extends ScannerCallable {
42
43
44
45
46
47 protected final byte[] locateStartRow;
48
49 @Deprecated
50 public ReversedScannerCallable(HConnection connection, TableName tableName, Scan scan,
51 ScanMetrics scanMetrics, byte[] locateStartRow) {
52 this(connection, tableName, scan, scanMetrics, locateStartRow, RpcControllerFactory
53 .instantiate(connection.getConfiguration()).newController());
54 }
55
56
57
58
59
60
61
62
63
64 public ReversedScannerCallable(HConnection connection, TableName tableName, Scan scan,
65 ScanMetrics scanMetrics, byte[] locateStartRow, PayloadCarryingRpcController rpcFactory) {
66 super(connection, tableName, scan, scanMetrics, rpcFactory);
67 this.locateStartRow = locateStartRow;
68 }
69
70
71
72
73
74 @Override
75 public void prepare(boolean reload) throws IOException {
76 if (!instantiated || reload) {
77 if (locateStartRow == null) {
78
79 this.location = connection.getRegionLocation(tableName, row, reload);
80 if (this.location == null) {
81 throw new IOException("Failed to find location, tableName="
82 + tableName + ", row=" + Bytes.toStringBinary(row) + ", reload="
83 + reload);
84 }
85 } else {
86
87
88 List<HRegionLocation> locatedRegions = locateRegionsInRange(
89 locateStartRow, row, reload);
90 if (locatedRegions.isEmpty()) {
91 throw new DoNotRetryIOException(
92 "Does hbase:meta exist hole? Couldn't get regions for the range from "
93 + Bytes.toStringBinary(locateStartRow) + " to "
94 + Bytes.toStringBinary(row));
95 }
96 this.location = locatedRegions.get(locatedRegions.size() - 1);
97 }
98 setStub(getConnection().getClient(getLocation().getServerName()));
99 checkIfRegionServerIsRemote();
100 instantiated = true;
101 }
102
103
104
105
106 if (reload && this.scanMetrics != null) {
107 this.scanMetrics.countOfRPCRetries.incrementAndGet();
108 if (isRegionServerRemote) {
109 this.scanMetrics.countOfRemoteRPCRetries.incrementAndGet();
110 }
111 }
112 }
113
114
115
116
117
118
119
120
121
122
123 private List<HRegionLocation> locateRegionsInRange(byte[] startKey,
124 byte[] endKey, boolean reload) throws IOException {
125 final boolean endKeyIsEndOfTable = Bytes.equals(endKey,
126 HConstants.EMPTY_END_ROW);
127 if ((Bytes.compareTo(startKey, endKey) > 0) && !endKeyIsEndOfTable) {
128 throw new IllegalArgumentException("Invalid range: "
129 + Bytes.toStringBinary(startKey) + " > "
130 + Bytes.toStringBinary(endKey));
131 }
132 List<HRegionLocation> regionList = new ArrayList<HRegionLocation>();
133 byte[] currentKey = startKey;
134 do {
135 HRegionLocation regionLocation = connection.getRegionLocation(tableName,
136 currentKey, reload);
137 if (regionLocation.getRegionInfo().containsRow(currentKey)) {
138 regionList.add(regionLocation);
139 } else {
140 throw new DoNotRetryIOException("Does hbase:meta exist hole? Locating row "
141 + Bytes.toStringBinary(currentKey) + " returns incorrect region "
142 + regionLocation.getRegionInfo());
143 }
144 currentKey = regionLocation.getRegionInfo().getEndKey();
145 } while (!Bytes.equals(currentKey, HConstants.EMPTY_END_ROW)
146 && (endKeyIsEndOfTable || Bytes.compareTo(currentKey, endKey) < 0));
147 return regionList;
148 }
149
150 }