Skip to content

Commit dab71ca

Browse files
HexyinUESTCHexyinUESTC
committed
add ObParams
1 parent 6966c4f commit dab71ca

File tree

4 files changed

+222
-67
lines changed

4 files changed

+222
-67
lines changed

‎pom.xml

Lines changed: 54 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@
4646

4747
<properties>
4848
<hadoop.version>1.2.1</hadoop.version>
49-
<hbase.version>1.3.6</hbase.version>
49+
<hbase.version>0.98.24-hadoop1</hbase.version>
5050
<java.source.version>1.8</java.source.version>
5151
<java.target.version>1.8</java.target.version>
5252
<junit.version>4.13.1</junit.version>
@@ -58,58 +58,6 @@
5858
</properties>
5959

6060
<dependencies>
61-
<dependency>
62-
<groupId>org.apache.hadoop</groupId>
63-
<artifactId>hadoop-core</artifactId>
64-
<version>${hadoop.version}</version>
65-
<exclusions>
66-
<exclusion>
67-
<artifactId>hsqldb</artifactId>
68-
<groupId>hsqldb</groupId>
69-
</exclusion>
70-
<exclusion>
71-
<artifactId>kfs</artifactId>
72-
<groupId>net.sf.kosmosfs</groupId>
73-
</exclusion>
74-
<exclusion>
75-
<artifactId>core</artifactId>
76-
<groupId>org.eclipse.jdt</groupId>
77-
</exclusion>
78-
<exclusion>
79-
<artifactId>jets3t</artifactId>
80-
<groupId>net.java.dev.jets3t</groupId>
81-
</exclusion>
82-
<exclusion>
83-
<artifactId>oro</artifactId>
84-
<groupId>oro</groupId>
85-
</exclusion>
86-
<exclusion>
87-
<artifactId>jasper-compiler</artifactId>
88-
<groupId>tomcat</groupId>
89-
</exclusion>
90-
<exclusion>
91-
<artifactId>jasper-runtime</artifactId>
92-
<groupId>tomcat</groupId>
93-
</exclusion>
94-
<exclusion>
95-
<artifactId>jackson-mapper-asl</artifactId>
96-
<groupId>org.codehaus.jackson</groupId>
97-
</exclusion>
98-
<exclusion>
99-
<artifactId>jettison</artifactId>
100-
<groupId>org.codehaus.jettison</groupId>
101-
</exclusion>
102-
<exclusion>
103-
<artifactId>commons-collections</artifactId>
104-
<groupId>commons-collections</groupId>
105-
</exclusion>
106-
</exclusions>
107-
</dependency>
108-
<dependency>
109-
<groupId>com.oceanbase</groupId>
110-
<artifactId>obkv-table-client</artifactId>
111-
<version>${table.client.version}</version>
112-
</dependency>
11361
<dependency>
11462
<groupId>org.apache.hbase</groupId>
11563
<artifactId>hbase-client</artifactId>
@@ -169,6 +117,58 @@
169117
</exclusion>
170118
</exclusions>
171119
</dependency>
120+
<dependency>
121+
<groupId>org.apache.hadoop</groupId>
122+
<artifactId>hadoop-core</artifactId>
123+
<version>${hadoop.version}</version>
124+
<exclusions>
125+
<exclusion>
126+
<artifactId>hsqldb</artifactId>
127+
<groupId>hsqldb</groupId>
128+
</exclusion>
129+
<exclusion>
130+
<artifactId>kfs</artifactId>
131+
<groupId>net.sf.kosmosfs</groupId>
132+
</exclusion>
133+
<exclusion>
134+
<artifactId>core</artifactId>
135+
<groupId>org.eclipse.jdt</groupId>
136+
</exclusion>
137+
<exclusion>
138+
<artifactId>jets3t</artifactId>
139+
<groupId>net.java.dev.jets3t</groupId>
140+
</exclusion>
141+
<exclusion>
142+
<artifactId>oro</artifactId>
143+
<groupId>oro</groupId>
144+
</exclusion>
145+
<exclusion>
146+
<artifactId>jasper-compiler</artifactId>
147+
<groupId>tomcat</groupId>
148+
</exclusion>
149+
<exclusion>
150+
<artifactId>jasper-runtime</artifactId>
151+
<groupId>tomcat</groupId>
152+
</exclusion>
153+
<exclusion>
154+
<artifactId>jackson-mapper-asl</artifactId>
155+
<groupId>org.codehaus.jackson</groupId>
156+
</exclusion>
157+
<exclusion>
158+
<artifactId>jettison</artifactId>
159+
<groupId>org.codehaus.jettison</groupId>
160+
</exclusion>
161+
<exclusion>
162+
<artifactId>commons-collections</artifactId>
163+
<groupId>commons-collections</groupId>
164+
</exclusion>
165+
</exclusions>
166+
</dependency>
167+
<dependency>
168+
<groupId>com.oceanbase</groupId>
169+
<artifactId>obkv-table-client</artifactId>
170+
<version>${table.client.version}</version>
171+
</dependency>
172172
<dependency>
173173
<groupId>org.slf4j</groupId>
174174
<artifactId>slf4j-api</artifactId>
@@ -404,4 +404,4 @@
404404
</plugins>
405405
</build>
406406
<profiles />
407-
</project>
407+
</project>

‎src/main/java/com/alipay/oceanbase/hbase/OHTable.java

Lines changed: 58 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,9 @@
2626
import com.alipay.oceanbase.hbase.util.ObTableClientManager;
2727
import com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory;
2828
import com.alipay.oceanbase.rpc.ObTableClient;
29+
import com.alipay.oceanbase.rpc.filter.ObHBaseParams;
30+
import com.alipay.oceanbase.rpc.filter.ObParamsBase;
31+
import com.alipay.oceanbase.rpc.filter.ObParams;
2932
import com.alipay.oceanbase.rpc.mutation.BatchOperation;
3033
import com.alipay.oceanbase.rpc.mutation.result.BatchOperationResult;
3134
import com.alipay.oceanbase.rpc.property.Property;
@@ -181,6 +184,8 @@ public class OHTable implements HTableInterface {
181184
*/
182185
private final Configuration configuration;
183186

187+
private int scannerTimeout;
188+
184189
/**
185190
* Creates an object to access a HBase table.
186191
* Shares oceanbase table obTableClient and other resources with other OHTable instances
@@ -203,6 +208,9 @@ public OHTable(Configuration configuration, String tableName) throws IOException
203208
DEFAULT_HBASE_HTABLE_PRIVATE_THREADS_MAX);
204209
this.keepAliveTime = configuration.getLong(HBASE_HTABLE_THREAD_KEEP_ALIVE_TIME,
205210
DEFAULT_HBASE_HTABLE_THREAD_KEEP_ALIVE_TIME);
211+
HBaseConfiguration.getInt(configuration, HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
212+
HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
213+
HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
206214
this.executePool = createDefaultThreadPoolExecutor(1, maxThreads, keepAliveTime);
207215
this.obTableClient = ObTableClientManager.getOrCreateObTableClient(configuration);
208216

@@ -364,8 +372,9 @@ public HTableDescriptor getTableDescriptor() {
364372
* @throws IOException e
365373
*/
366374
public boolean exists(Get get) throws IOException {
375+
get.setCheckExistenceOnly(true);
367376
Result r = get(get);
368-
return !r.isEmpty();
377+
return r.getExists();
369378
}
370379

371380
@Override
@@ -380,8 +389,8 @@ public boolean[] existsAll(List<Get> list) throws IOException {
380389
// todo: Optimize after CheckExistenceOnly is finished
381390
Result[] r = get(list);
382391
boolean[] ret = new boolean[r.length];
383-
for (int i = 0; i < r.length; ++i){
384-
ret[i] = !r[i].isEmpty();
392+
for (int i = 0; i < list.size(); ++i) {
393+
ret[i] = exists(list.get(i));
385394
}
386395
return ret;
387396
}
@@ -470,10 +479,20 @@ public Result call() throws IOException {
470479
get.getMaxVersions(), null);
471480
obTableQuery = buildObTableQuery(filter, get.getRow(), true, get.getRow(),
472481
true, -1);
482+
obTableQuery.setObParams(buildObHBaseParams(null, get));
473483
request = buildObTableQueryRequest(obTableQuery, getTargetTableName(tableNameString));
474484

475485
clientQueryStreamResult = (ObTableClientQueryStreamResult) obTableClient
476486
.execute(request);
487+
if (get.isCheckExistenceOnly() ) {
488+
Result result = new Result();
489+
if (clientQueryStreamResult.getCacheRows().size() != 0) {
490+
result.setExists(true);
491+
} else {
492+
result.setExists(false);
493+
}
494+
return result;
495+
}
477496
getKeyValueFromResult(clientQueryStreamResult, keyValueList, true, family);
478497
} else {
479498
for (Map.Entry<byte[], NavigableSet<byte[]>> entry : get.getFamilyMap()
@@ -492,10 +511,20 @@ public Result call() throws IOException {
492511
get.getRow(), true, -1);
493512
}
494513

514+
obTableQuery.setObParams(buildObHBaseParams(null, get));
495515
request = buildObTableQueryRequest(obTableQuery,
496516
getTargetTableName(tableNameString, Bytes.toString(family)));
497517
clientQueryStreamResult = (ObTableClientQueryStreamResult) obTableClient
498518
.execute(request);
519+
if (get.isCheckExistenceOnly() ) {
520+
Result result = new Result();
521+
if (clientQueryStreamResult.getCacheRows().size() != 0) {
522+
result.setExists(true);
523+
} else {
524+
result.setExists(false);
525+
}
526+
return result;
527+
}
499528
getKeyValueFromResult(clientQueryStreamResult, keyValueList, false,
500529
family);
501530
}
@@ -555,17 +584,18 @@ public ResultScanner call() throws IOException {
555584
scan.getMaxVersions(), null);
556585
if (scan.isReversed()) {
557586
obTableQuery = buildObTableQuery(filter, scan.getStopRow(), false,
558-
scan.getStartRow(), true, scan.getBatch());
587+
scan.getStartRow(), true, scan.getCaching());
559588
} else {
560589
obTableQuery = buildObTableQuery(filter, scan.getStartRow(), true,
561-
scan.getStopRow(), false, scan.getBatch());
590+
scan.getStopRow(), false, scan.getCaching());
562591
}
563592
if (scan.isReversed()) { // reverse scan 时设置为逆序
564593
obTableQuery.setScanOrder(ObScanOrder.Reverse);
565594
}
566595
obTableQuery.setMaxResultSize(scan.getMaxResultSize() > 0 ? scan.getMaxResultSize() : conf.getLong(
567596
HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
568597
HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE));
598+
obTableQuery.setObParams(buildObHBaseParams(scan, null));
569599
request = buildObTableQueryAsyncRequest(obTableQuery, getTargetTableName(tableNameString));
570600
clientQueryAsyncStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
571601
.execute(request);
@@ -579,20 +609,19 @@ public ResultScanner call() throws IOException {
579609
scan.getMaxVersions(), entry.getValue());
580610
if (scan.isReversed()) {
581611
obTableQuery = buildObTableQuery(filter, scan.getStopRow(), false,
582-
scan.getStartRow(), true, scan.getBatch());
612+
scan.getStartRow(), true, scan.getCaching());
583613
} else {
584614
obTableQuery = buildObTableQuery(filter, scan.getStartRow(), true,
585-
scan.getStopRow(), false, scan.getBatch());
615+
scan.getStopRow(), false, scan.getCaching());
586616
}
587617
if (scan.isReversed()) { // reverse scan 时设置为逆序
588618
obTableQuery.setScanOrder(ObScanOrder.Reverse);
589619
}
590620

591-
// no support set maxResultSize.
592621
obTableQuery.setMaxResultSize(scan.getMaxResultSize() > 0 ? scan.getMaxResultSize() : conf.getLong(
593622
HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
594623
HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE));
595-
624+
obTableQuery.setObParams(buildObHBaseParams(scan, null));
596625
request = buildObTableQueryAsyncRequest(obTableQuery,
597626
getTargetTableName(tableNameString, Bytes.toString(family)));
598627
clientQueryAsyncStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
@@ -614,6 +643,24 @@ public ResultScanner call() throws IOException {
614643
return executeServerCallable(serverCallable);
615644
}
616645

646+
public ObParams buildObHBaseParams(Scan scan, Get get) {
647+
ObParams obParams = new ObParams();
648+
ObHBaseParams obHBaseParams = new ObHBaseParams();
649+
if (scan != null) {
650+
obHBaseParams.setBatch(scan.getBatch());
651+
obHBaseParams.setCallTimeout(scannerTimeout);
652+
obHBaseParams.setRaw(scan.isRaw());
653+
obHBaseParams.setCacheBlock(scan.isGetScan());
654+
obHBaseParams.setAllowPartialResults(scan.getAllowPartialResults());
655+
}
656+
if (get != null) {
657+
obHBaseParams.setCheckExistenceOnly(get.isCheckExistenceOnly());
658+
obHBaseParams.setCacheBlock(get.getCacheBlocks());
659+
}
660+
obParams.setObParamsBase(obHBaseParams);
661+
return obParams;
662+
}
663+
617664
public ResultScanner getScanner(final byte[] family) throws IOException {
618665
Scan scan = new Scan();
619666
scan.addFamily(family);
@@ -1510,10 +1557,10 @@ public void refreshTableEntry(String familyString, boolean hasTestLoad) throws E
15101557
return;
15111558
}
15121559
this.obTableClient.getOrRefreshTableEntry(
1513-
getNormalTargetTableName(tableNameString, familyString), true, true);
1560+
getNormalTargetTableName(tableNameString, familyString), true, true, false);
15141561
if (hasTestLoad) {
15151562
this.obTableClient.getOrRefreshTableEntry(
1516-
getTestLoadTargetTableName(tableNameString, familyString), true, true);
1563+
getTestLoadTargetTableName(tableNameString, familyString), true, true, false);
15171564
}
15181565
}
15191566

‎src/main/java/com/alipay/oceanbase/hbase/result/ClientStreamScanner.java

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -106,8 +106,10 @@ public Result next() throws IOException {
106106
KeyValue startKeyValue = new KeyValue(sk, family, sq, st, sv);
107107
List<KeyValue> keyValues = new ArrayList<KeyValue>();
108108
keyValues.add(startKeyValue);
109-
110-
while (streamNext = streamResult.next()) {
109+
//
110+
int size = streamResult.getCacheRows().size();
111+
int current = 0;
112+
while (streamNext = streamResult.next() && current < size){
111113
List<ObObj> row = streamResult.getRow();
112114
if (this.isTableGroup) {
113115
// split family and qualifier
@@ -124,6 +126,7 @@ public Result next() throws IOException {
124126
if (Arrays.equals(sk, k)) {
125127
// when rowKey is equal to the previous rowKey ,merge the result into the same result
126128
keyValues.add(new KeyValue(k, family, q, t, v));
129+
current++;
127130
} else {
128131
break;
129132
}

0 commit comments

Comments
 (0)