Skip to content

Commit d314d46

Browse files
stuBirdFlystuBirdFly
committed
fix bug when get set isClosestRow true
1 parent 21a3ee3 commit d314d46

13 files changed

+173
-143
lines changed

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

Lines changed: 69 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@
3535
import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.*;
3636
import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.syncquery.ObTableQueryAsyncRequest;
3737
import com.alipay.oceanbase.rpc.stream.ObTableClientQueryAsyncStreamResult;
38-
import com.alipay.oceanbase.rpc.stream.ObTableClientQueryStreamResult;
3938
import com.alipay.oceanbase.rpc.table.ObHBaseParams;
4039
import com.alipay.oceanbase.rpc.table.ObKVParams;
4140
import com.alipay.sofa.common.thread.SofaThreadPoolExecutor;
@@ -49,10 +48,7 @@
4948
import org.apache.hadoop.hbase.*;
5049
import org.apache.hadoop.hbase.client.*;
5150
import org.apache.hadoop.hbase.client.coprocessor.Batch;
52-
import org.apache.hadoop.hbase.filter.CompareFilter;
53-
import org.apache.hadoop.hbase.filter.Filter;
54-
import org.apache.hadoop.hbase.filter.FilterList;
55-
import org.apache.hadoop.hbase.filter.PageFilter;
51+
import org.apache.hadoop.hbase.filter.*;
5652
import org.apache.hadoop.hbase.io.TimeRange;
5753
import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
5854
import org.apache.hadoop.hbase.util.Bytes;
@@ -401,8 +397,8 @@ public HTableDescriptor getTableDescriptor() {
401397
*/
402398
@Override
403399
public boolean exists(Get get) throws IOException {
404-
Result r = get(get);
405-
return !r.isEmpty();
400+
get.setCheckExistenceOnly(true);
401+
return this.get(get).getExists();
406402
}
407403

408404
@Override
@@ -455,27 +451,51 @@ public <R> Object[] batchCallback(List<? extends Row> actions, Batch.Callback<R>
455451
throw new FeatureNotSupportedException("not supported yet'");
456452
}
457453

458-
private void getKeyValueFromResult(AbstractQueryStreamResult clientQueryStreamResult,
459-
List<KeyValue> keyValueList, boolean isTableGroup,
460-
byte[] family) throws Exception {
454+
public static int compareByteArray(byte[] bt1, byte[] bt2) {
455+
int minLength = Math.min(bt1.length, bt2.length);
456+
for (int i = 0; i < minLength; i++) {
457+
if (bt1[i] != bt2[i]) {
458+
return bt1[i] - bt2[i];
459+
}
460+
}
461+
return bt1.length - bt2.length;
462+
}
463+
464+
private void getMaxRowFromResult(AbstractQueryStreamResult clientQueryStreamResult,
465+
List<KeyValue> keyValueList, boolean isTableGroup,
466+
byte[] family) throws Exception {
461467
byte[][] familyAndQualifier = new byte[2][];
468+
KeyValue kv = null;
462469
while (clientQueryStreamResult.next()) {
463470
List<ObObj> row = clientQueryStreamResult.getRow();
464-
if (isTableGroup) {
465-
// split family and qualifier
466-
familyAndQualifier = OHBaseFuncUtils.extractFamilyFromQualifier((byte[]) row.get(1)
467-
.getValue());
471+
if (row.isEmpty()) {
472+
// Currently, checkExistOnly is set, and if the row exists, it returns an empty row.
473+
keyValueList.add(new KeyValue());
474+
return;
468475
} else {
469-
familyAndQualifier[0] = family;
470-
familyAndQualifier[1] = (byte[]) row.get(1).getValue();
476+
if (kv == null
477+
|| compareByteArray(kv.getRow(), (byte[]) row.get(0).getValue()) <= 0) {
478+
if (kv != null
479+
&& compareByteArray(kv.getRow(), (byte[]) row.get(0).getValue()) != 0) {
480+
keyValueList.clear();
481+
}
482+
if (isTableGroup) {
483+
// split family and qualifier
484+
familyAndQualifier = OHBaseFuncUtils
485+
.extractFamilyFromQualifier((byte[]) row.get(1).getValue());
486+
} else {
487+
familyAndQualifier[0] = family;
488+
familyAndQualifier[1] = (byte[]) row.get(1).getValue();
489+
}
490+
kv = new KeyValue((byte[]) row.get(0).getValue(),//K
491+
familyAndQualifier[0], // family
492+
familyAndQualifier[1], // qualifiermat
493+
(Long) row.get(2).getValue(), // T
494+
(byte[]) row.get(3).getValue()// V
495+
);
496+
keyValueList.add(kv);
497+
}
471498
}
472-
KeyValue kv = new KeyValue((byte[]) row.get(0).getValue(),//K
473-
familyAndQualifier[0], // family
474-
familyAndQualifier[1], // qualifier
475-
(Long) row.get(2).getValue(), // T
476-
(byte[]) row.get(3).getValue()// V
477-
);
478-
keyValueList.add(kv);
479499
}
480500
}
481501

@@ -491,7 +511,8 @@ private String getTargetTableName(String tableNameString) {
491511
// To enable the server to identify the column family to which a qualifier belongs,
492512
// the client writes the column family name into the qualifier.
493513
// The server then parses this information to determine the table that needs to be operated on.
494-
private void processColumnFilters(NavigableSet<byte[]> columnFilters, Map<byte[], NavigableSet<byte[]>> familyMap) {
514+
private void processColumnFilters(NavigableSet<byte[]> columnFilters,
515+
Map<byte[], NavigableSet<byte[]>> familyMap) {
495516
for (Map.Entry<byte[], NavigableSet<byte[]>> entry : familyMap.entrySet()) {
496517
if (entry.getValue() != null) {
497518
for (byte[] columnName : entry.getValue()) {
@@ -537,7 +558,7 @@ public Result call() throws IOException {
537558

538559
clientQueryStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
539560
.execute(request);
540-
getKeyValueFromResult(clientQueryStreamResult, keyValueList, true, family);
561+
getMaxRowFromResult(clientQueryStreamResult, keyValueList, true, family);
541562
} else {
542563
for (Map.Entry<byte[], NavigableSet<byte[]>> entry : get.getFamilyMap()
543564
.entrySet()) {
@@ -548,7 +569,7 @@ public Result call() throws IOException {
548569
configuration));
549570
clientQueryStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
550571
.execute(request);
551-
getKeyValueFromResult(clientQueryStreamResult, keyValueList, false,
572+
getMaxRowFromResult(clientQueryStreamResult, keyValueList, false,
552573
family);
553574
}
554575
}
@@ -558,6 +579,9 @@ public Result call() throws IOException {
558579
throw new IOException("query table:" + tableNameString + " family "
559580
+ Bytes.toString(family) + " error.", e);
560581
}
582+
if (get.isCheckExistenceOnly()) {
583+
return Result.create(null, !keyValueList.isEmpty());
584+
}
561585
return new Result(keyValueList);
562586
}
563587
};
@@ -1110,7 +1134,7 @@ public void flushCommits() throws IOException {
11101134
// Bypass logic: directly construct BatchOperation for puts with family map size > 1
11111135
try {
11121136
BatchOperation batch = buildBatchOperation(this.tableNameString,
1113-
innerFamilyMap, false, null);
1137+
innerFamilyMap, false, null);
11141138
BatchOperationResult results = batch.execute();
11151139

11161140
boolean hasError = results.hasError();
@@ -1120,20 +1144,20 @@ public void flushCommits() throws IOException {
11201144
}
11211145
} catch (Exception e) {
11221146
logger.error(LCD.convert("01-00008"), tableNameString, null, autoFlush,
1123-
writeBuffer.size(), e);
1147+
writeBuffer.size(), e);
11241148
throw new IOException("put table " + tableNameString + " error codes "
1125-
+ null + "auto flush " + autoFlush
1126-
+ " current buffer size " + writeBuffer.size(), e);
1149+
+ null + "auto flush " + autoFlush
1150+
+ " current buffer size " + writeBuffer.size(), e);
11271151
}
11281152
} else {
11291153
// Existing logic for puts with family map size = 1
11301154
for (Map.Entry<byte[], List<KeyValue>> entry : innerFamilyMap.entrySet()) {
11311155
String family = Bytes.toString(entry.getKey());
11321156
Pair<List<Integer>, List<KeyValue>> keyValueWithIndex = familyMap
1133-
.get(family);
1157+
.get(family);
11341158
if (keyValueWithIndex == null) {
11351159
keyValueWithIndex = new Pair<List<Integer>, List<KeyValue>>(
1136-
new ArrayList<Integer>(), new ArrayList<KeyValue>());
1160+
new ArrayList<Integer>(), new ArrayList<KeyValue>());
11371161
familyMap.put(family, keyValueWithIndex);
11381162
}
11391163
keyValueWithIndex.getFirst().add(i);
@@ -1414,7 +1438,8 @@ private static String getTestLoadTargetTableName(String tableNameString, String
14141438
}
14151439

14161440
private ObHTableFilter buildObHTableFilter(Filter filter, TimeRange timeRange, int maxVersion,
1417-
Collection<byte[]> columnQualifiers) throws IOException {
1441+
Collection<byte[]> columnQualifiers)
1442+
throws IOException {
14181443
ObHTableFilter obHTableFilter = new ObHTableFilter();
14191444

14201445
if (filter != null) {
@@ -1440,7 +1465,9 @@ private ObHTableFilter buildObHTableFilter(Filter filter, TimeRange timeRange, i
14401465
return obHTableFilter;
14411466
}
14421467

1443-
private byte[] buildCheckAndMutateFilterString(byte[] family, byte[] qualifier, CompareFilter.CompareOp compareOp, byte[] value) throws IOException {
1468+
private byte[] buildCheckAndMutateFilterString(byte[] family, byte[] qualifier,
1469+
CompareFilter.CompareOp compareOp, byte[] value)
1470+
throws IOException {
14441471
ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
14451472
byteStream.write("CheckAndMutateFilter(".getBytes());
14461473
byteStream.write(HBaseFilterUtils.toParseableByteArray(compareOp));
@@ -1545,7 +1572,8 @@ private ObTableQuery buildObTableQuery(ObHTableFilter filter, final Scan scan) {
15451572
return obTableQuery;
15461573
}
15471574

1548-
private ObTableQuery buildObTableQuery(final Get get, Collection<byte[]> columnQualifiers) throws IOException {
1575+
private ObTableQuery buildObTableQuery(final Get get, Collection<byte[]> columnQualifiers)
1576+
throws IOException {
15491577
ObTableQuery obTableQuery;
15501578
if (get.isClosestRowBefore()) {
15511579
PageFilter pageFilter = new PageFilter(1);
@@ -1593,11 +1621,11 @@ private ObTableBatchOperation buildObTableBatchOperation(Map<byte[], List<KeyVal
15931621
for (KeyValue kv : keyValueList) {
15941622
if (qualifiers != null) {
15951623
qualifiers
1596-
.add((Bytes.toString(family) + "." + Bytes.toString(kv.getQualifier()))
1597-
.getBytes());
1624+
.add((Bytes.toString(family) + "." + Bytes.toString(kv.getQualifier()))
1625+
.getBytes());
15981626
}
15991627
KeyValue new_kv = modifyQualifier(kv,
1600-
(Bytes.toString(family) + "." + Bytes.toString(kv.getQualifier())).getBytes());
1628+
(Bytes.toString(family) + "." + Bytes.toString(kv.getQualifier())).getBytes());
16011629
batch.addTableOperation(buildObTableOperation(new_kv, putToAppend));
16021630
}
16031631
}
@@ -1636,6 +1664,7 @@ private com.alipay.oceanbase.rpc.mutation.Mutation buildMutation(KeyValue kv,
16361664
throw new IllegalArgumentException("illegal mutation type " + kvType);
16371665
}
16381666
}
1667+
16391668
private KeyValue modifyQualifier(KeyValue original, byte[] newQualifier) {
16401669
// Extract existing components
16411670
byte[] row = original.getRow();
@@ -1645,7 +1674,7 @@ private KeyValue modifyQualifier(KeyValue original, byte[] newQualifier) {
16451674
byte type = original.getTypeByte();
16461675
// Create a new KeyValue with the modified qualifier
16471676
return new KeyValue(row, family, newQualifier, timestamp, KeyValue.Type.codeToType(type),
1648-
value);
1677+
value);
16491678
}
16501679

16511680
private BatchOperation buildBatchOperation(String tableName,
@@ -1661,7 +1690,7 @@ private BatchOperation buildBatchOperation(String tableName,
16611690
qualifiers.add(kv.getQualifier());
16621691
}
16631692
KeyValue new_kv = modifyQualifier(kv,
1664-
(Bytes.toString(family) + "." + Bytes.toString(kv.getQualifier())).getBytes());
1693+
(Bytes.toString(family) + "." + Bytes.toString(kv.getQualifier())).getBytes());
16651694
batch.addOperation(buildMutation(new_kv, putToAppend));
16661695
}
16671696
}
@@ -1670,7 +1699,6 @@ private BatchOperation buildBatchOperation(String tableName,
16701699
return batch;
16711700
}
16721701

1673-
16741702
private BatchOperation buildBatchOperation(String tableName, List<KeyValue> keyValueList,
16751703
boolean putToAppend, List<byte[]> qualifiers) {
16761704
BatchOperation batch = obTableClient.batchOperation(tableName);
@@ -1771,7 +1799,6 @@ public static void checkFamilyViolation(Collection<byte[]> families, boolean che
17711799
}
17721800
}
17731801

1774-
17751802
// This method is currently only used for append and increment operations.
17761803
// It restricts these two methods to use multi-column family operations.
17771804
// Note: After completing operations on multiple column families, they are deleted using the method described above.

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
package com.alipay.oceanbase.hbase;
1919

2020
import com.alipay.oceanbase.hbase.core.Lifecycle;
21-
import com.alipay.oceanbase.hbase.exception.FeatureNotSupportedException;
2221
import com.google.protobuf.Descriptors;
2322
import com.google.protobuf.Message;
2423
import com.google.protobuf.Service;

0 commit comments

Comments
 (0)