Skip to content

Commit 2e4c625

Browse files
maochongxinmaochongxin
committed
impl region locator
1 parent 1948c10 commit 2e4c625

File tree

6 files changed

+350
-26
lines changed

6 files changed

+350
-26
lines changed

‎src/main/java/com/alipay/oceanbase/hbase/execute/AbstractObTableMetaExecutor.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ public T execute(ObTableClient client, ObTableMetaRequest request) throws IOExce
1515
if (request.getMetaType() != getMetaType()) {
1616
throw new IOException("Invalid meta type, expected " + getMetaType());
1717
}
18-
ObTable table = client.randomTable();
18+
ObTable table = client.getRandomTable();
1919
ObTableMetaResponse response;
2020
try {
2121
response = (ObTableMetaResponse) client.executeWithRetry(

‎src/main/java/com/alipay/oceanbase/hbase/util/OHConnectionImpl.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -149,8 +149,9 @@ public BufferedMutator getBufferedMutator(BufferedMutatorParams params, OHTable
149149

150150
@Override
151151
public RegionLocator getRegionLocator(TableName tableName) throws IOException {
152-
ObTableClient obTableClient = ObTableClientManager.getOrCreateObTableClient(null/*args*/);
153-
OHRegionLocatorExecutor executor = new OHRegionLocatorExecutor(obTableClient);
152+
OHConnectionConfiguration connectionConfig = getOHConnectionConfiguration();
153+
ObTableClient obTableClient = ObTableClientManager.getOrCreateObTableClient(connectionConfig);
154+
OHRegionLocatorExecutor executor = new OHRegionLocatorExecutor(tableName.toString(), obTableClient);
154155
return executor.getRegionLocator(String.valueOf(tableName));
155156
}
156157

‎src/main/java/com/alipay/oceanbase/hbase/util/OHRegionLocator.java

Lines changed: 30 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,23 +11,44 @@
1111
import java.util.List;
1212

1313
public class OHRegionLocator implements RegionLocator {
14-
public OHRegionLocator(byte[][] startKeys, byte[][] endKeys) {
15-
14+
private byte[][] startKeys;
15+
private byte[][] endKeys;
16+
private TableName tableName;
17+
18+
private List<HRegionLocation> regionLocations;
19+
20+
public OHRegionLocator(byte[][] startKeys, byte[][] endKeys, List<HRegionLocation> regionLocations) {
21+
this.startKeys = startKeys;
22+
this.endKeys = endKeys;
23+
this.regionLocations = regionLocations;
1624
}
1725

1826
@Override
1927
public HRegionLocation getRegionLocation(byte[] bytes) throws IOException {
28+
// check if bytes is in the range of startKeys and endKeys
29+
for (HRegionLocation regionLocation : regionLocations) {
30+
if (regionLocation.getRegionInfo().containsRow(bytes)) {
31+
return regionLocation;
32+
}
33+
}
2034
return null;
2135
}
2236

2337
@Override
2438
public HRegionLocation getRegionLocation(byte[] bytes, boolean b) throws IOException {
25-
return null;
39+
if (b) {
40+
OHRegionLocatorExecutor executor = new OHRegionLocatorExecutor(tableName.toString(), tableClient);
41+
RegionLocator location = executor.getRegionLocator(tableName.toString());
42+
this.startKeys = location.getStartKeys();
43+
this.endKeys = location.getEndKeys();
44+
this.regionLocations = location.getAllRegionLocations();
45+
}
46+
return getRegionLocation(bytes);
2647
}
2748

2849
@Override
2950
public List<HRegionLocation> getAllRegionLocations() throws IOException {
30-
return Collections.emptyList();
51+
return regionLocations;
3152
}
3253

3354
/**
@@ -40,7 +61,7 @@ public List<HRegionLocation> getAllRegionLocations() throws IOException {
4061
*/
4162
@Override
4263
public byte[][] getStartKeys() throws IOException {
43-
return null;
64+
return startKeys;
4465
}
4566

4667
/**
@@ -53,7 +74,7 @@ public byte[][] getStartKeys() throws IOException {
5374
*/
5475
@Override
5576
public byte[][] getEndKeys() throws IOException {
56-
return null;
77+
return endKeys;
5778
}
5879

5980
/**
@@ -67,18 +88,18 @@ public byte[][] getEndKeys() throws IOException {
6788
*/
6889
@Override
6990
public Pair<byte[][], byte[][]> getStartEndKeys() throws IOException {
70-
return null;
91+
return Pair.newPair(startKeys, endKeys);
7192
}
7293

7394
@Override
7495
public TableName getName() {
75-
return null;
96+
return tableName;
7697
}
7798

7899
private ObTableClient tableClient;
79100

80101
@Override
81102
public void close() throws IOException {
82-
103+
return;
83104
}
84105
}

‎src/main/java/com/alipay/oceanbase/hbase/util/OHRegionLocatorExecutor.java

Lines changed: 137 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,28 @@
11
package com.alipay.oceanbase.hbase.util;
22

33
import com.alibaba.fastjson.JSON;
4+
import com.alibaba.fastjson.JSONObject;
45
import com.alipay.oceanbase.hbase.execute.AbstractObTableMetaExecutor;
56
import com.alipay.oceanbase.rpc.ObTableClient;
7+
import com.alipay.oceanbase.rpc.constant.Constants;
68
import com.alipay.oceanbase.rpc.exception.ObTableException;
79
import com.alipay.oceanbase.rpc.location.model.TableEntry;
810
import com.alipay.oceanbase.rpc.meta.ObTableMetaRequest;
911
import com.alipay.oceanbase.rpc.meta.ObTableMetaResponse;
1012
import com.alipay.oceanbase.rpc.meta.ObTableRpcMetaType;
13+
import org.apache.hadoop.hbase.*;
1114

1215
import java.io.IOException;
13-
import java.util.HashMap;
14-
import java.util.Map;
16+
import java.util.*;
17+
import java.util.stream.Collectors;
18+
import java.util.stream.IntStream;
1519

1620
public class OHRegionLocatorExecutor extends AbstractObTableMetaExecutor<OHRegionLocator> {
17-
OHRegionLocatorExecutor(ObTableClient client) {
21+
private final String tableName;
22+
private final ObTableClient client;
23+
24+
OHRegionLocatorExecutor(String tableName, ObTableClient client) {
25+
this.tableName = tableName;
1826
this.client = client;
1927
}
2028

@@ -26,24 +34,140 @@ public ObTableRpcMetaType getMetaType() {
2634
@Override
2735
public OHRegionLocator parse(ObTableMetaResponse response) throws IOException {
2836
try {
29-
String jsonData = response.getData();
30-
// process json
31-
return new OHRegionLocator(null, null);
37+
final String jsonData = response.getData();
38+
final JSONObject jsonMap = Optional.<JSONObject>ofNullable(JSON.parseObject(jsonData))
39+
.orElseThrow(() -> new IOException("jsonMap is null"));
40+
41+
final List<Object> partitions = Optional.<List<Object>>ofNullable(jsonMap.getJSONArray("partitions"))
42+
.orElseThrow(() -> new IOException("partitions is null"));
43+
44+
final List<Object> tableIdDict = Optional.<List<Object>>ofNullable(jsonMap.getJSONArray("table_id_dict"))
45+
.orElseThrow(() -> new IOException("tableIdDict is null"));
46+
final List<Object> replicaDict = Optional.<List<Object>>ofNullable(jsonMap.getJSONArray("replica_dict"))
47+
.orElseThrow(() -> new IOException("replicaDict is null"));
48+
49+
final boolean isHashLikePartition = partitions.stream()
50+
.map(obj -> (List<Object>) obj)
51+
.filter(partition -> {
52+
if (partition.size() <= 3) {
53+
throw new IllegalArgumentException("partition size is not 3");
54+
}
55+
return true;
56+
})
57+
.allMatch(partition -> {
58+
final byte[] highBound = partition.get(2).toString().getBytes();
59+
return Arrays.equals(highBound, Constants.EMPTY_STRING.getBytes());
60+
});
61+
return isHashLikePartition ?
62+
createHashPartitionLocator(tableIdDict, replicaDict, partitions) :
63+
createRangePartitionLocator(tableIdDict, replicaDict, partitions);
3264
} catch (IllegalArgumentException e) {
33-
throw new IOException("msg", e);
65+
throw new IOException("Invalid partition data: " + e.getMessage(), e);
66+
}
67+
}
68+
69+
/**
70+
* Creates a region locator for range partitions
71+
* @param tableIdDict table ID dictionary
72+
* @param replicaDict replica dictionary
73+
* @param partitions list of partition data
74+
* @return OHRegionLocator for range partitions
75+
*/
76+
private OHRegionLocator createRangePartitionLocator(
77+
final List<Object> tableIdDict,
78+
final List<Object> replicaDict,
79+
final List<Object> partitions) {
80+
final int partitionCount = partitions.size();
81+
final int regionCount = partitionCount + 1;
82+
final byte[][] startKeys = new byte[regionCount][];
83+
final byte[][] endKeys = new byte[regionCount][];
84+
85+
for (int i = 0; i < regionCount; i++) {
86+
if (i == 0) {
87+
startKeys[i] = HConstants.EMPTY_BYTE_ARRAY;
88+
endKeys[i] = ((List<Object>) partitions.get(i)).get(2).toString().getBytes();
89+
} else if (i == regionCount - 1) {
90+
startKeys[i] = ((List<Object>) partitions.get(i - 1)).get(2).toString().getBytes();
91+
endKeys[i] = HConstants.EMPTY_BYTE_ARRAY;
92+
} else {
93+
startKeys[i] = ((List<Object>) partitions.get(i - 1)).get(2).toString().getBytes();
94+
endKeys[i] = ((List<Object>) partitions.get(i)).get(2).toString().getBytes();
95+
}
3496
}
97+
98+
// Create region locations for all regions
99+
final List<HRegionLocation> regionLocations = IntStream.range(0, regionCount)
100+
.mapToObj(i -> {
101+
final List<Object> partition = (List<Object>) partitions.get(Math.min(i, partitionCount - 1));
102+
final int replicationIdx = (int) partition.get(3);
103+
final List<Object> hostInfo = (List<Object>) replicaDict.get(replicationIdx);
104+
105+
final ServerName serverName = ServerName.valueOf(
106+
hostInfo.get(0).toString(),
107+
(int) hostInfo.get(1),
108+
i
109+
);
110+
final HRegionInfo regionInfo = new HRegionInfo(
111+
TableName.valueOf(tableName),
112+
startKeys[i],
113+
endKeys[i]
114+
);
115+
return new HRegionLocation(regionInfo, serverName, i);
116+
})
117+
.collect(Collectors.toList());
118+
119+
return new OHRegionLocator(startKeys, endKeys, regionLocations);
120+
}
121+
122+
/**
123+
* Creates a region locator for hash partitions
124+
* @param tableIdDict table ID dictionary
125+
* @param replicaDict replica dictionary
126+
* @param partitions list of partition data
127+
* @return OHRegionLocator for hash partitions
128+
*/
129+
private OHRegionLocator createHashPartitionLocator(
130+
final List<Object> tableIdDict,
131+
final List<Object> replicaDict,
132+
final List<Object> partitions) {
133+
134+
final byte[][] startKeys = new byte[1][];
135+
final byte[][] endKeys = new byte[1][];
136+
startKeys[0] = HConstants.EMPTY_BYTE_ARRAY;
137+
endKeys[0] = HConstants.EMPTY_BYTE_ARRAY;
138+
139+
final List<HRegionLocation> regionLocations = IntStream.range(0, partitions.size())
140+
.mapToObj(i -> {
141+
final List<Object> partition = (List<Object>) partitions.get(i);
142+
final int replicationIdx = (int) partition.get(3);
143+
final List<Object> hostInfo = (List<Object>) replicaDict.get(replicationIdx);
144+
145+
final ServerName serverName = ServerName.valueOf(
146+
hostInfo.get(0).toString(),
147+
(int) hostInfo.get(1),
148+
i
149+
);
150+
final HRegionInfo regionInfo = new HRegionInfo(
151+
TableName.valueOf(tableName),
152+
startKeys[0],
153+
endKeys[0]
154+
);
155+
return new HRegionLocation(regionInfo, serverName, i);
156+
})
157+
.collect(Collectors.toList());
158+
159+
return new OHRegionLocator(startKeys, endKeys, regionLocations);
35160
}
36161

37-
public OHRegionLocator getRegionLocator(String tableName) throws IOException {
38-
ObTableMetaRequest request = new ObTableMetaRequest();
162+
public OHRegionLocator getRegionLocator(final String tableName) throws IOException {
163+
final ObTableMetaRequest request = new ObTableMetaRequest();
39164
request.setMetaType(getMetaType());
40-
Map<String, Object> requestData = new HashMap<>();
165+
final Map<String, String> requestData = new HashMap<>();
41166
requestData.put("table_name", tableName);
42-
String jsonData = JSON.toJSONString(requestData);
167+
168+
final String jsonData = JSON.toJSONString(requestData);
43169
request.setData(jsonData);
44170

45171
return execute(client, request);
46172
}
47-
48-
private final ObTableClient client;
49173
}

0 commit comments

Comments
 (0)