Skip to content

update obkv-hbase example #147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
stuBirdFly merged 1 commit into from
Feb 7, 2025
Merged

update obkv-hbase example #147

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion example/simple-hbase-demo/src/main/java/com/oceanbase/example/SimpleHBaseClientDemo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,10 @@ public static void simpleTest() throws Exception {
Get get = new Get(rowKey);
get.addColumn(family, column);
Result r = hTable.get(get);
System.out.printf("column1: " + r.getColumn(family, column));
if (!r.isEmpty()) {
Cell cell = r.rawCells()[0];
System.out.printf("column1: " + CellUtil.cloneQualifier(r));
}

// 4. close
hTable.close();
Expand Down
74 changes: 32 additions & 42 deletions src/main/java/com/alipay/oceanbase/hbase/OHTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -801,7 +801,7 @@ private String getTargetTableName(List<? extends Row> actions) {
throw new FeatureNotSupportedException("not supported yet'");
} else {
Set<byte[]> familySet = null;
if (action instanceof Get){
if (action instanceof Get) {
Get get = (Get) action;
familySet = get.familySet();
} else {
Expand Down Expand Up @@ -921,7 +921,7 @@ private void processColumnFilters(NavigableSet<byte[]> columnFilters,

@Override
public Result get(final Get get) throws IOException {
if (get.getFamilyMap().keySet() == null || get.getFamilyMap().keySet().isEmpty()) {
if (get.getFamilyMap().keySet().isEmpty()) {
// check nothing, use table group;
} else {
checkFamilyViolation(get.getFamilyMap().keySet(), false);
Expand All @@ -934,8 +934,7 @@ public Result call() throws IOException {
byte[] family = new byte[] {};
ObTableQuery obTableQuery;
try {
if (get.getFamilyMap().keySet() == null
|| get.getFamilyMap().keySet().isEmpty()
if (get.getFamilyMap().keySet().isEmpty()
|| get.getFamilyMap().size() > 1) {
// In a Get operation where the family map is greater than 1 or equal to 0,
// we handle this by appending the column family to the qualifier on the client side.
Expand Down Expand Up @@ -1024,8 +1023,7 @@ public ResultScanner call() throws IOException {
ObTableQuery obTableQuery;
ObHTableFilter filter;
try {
if (scan.getFamilyMap().keySet() == null
|| scan.getFamilyMap().keySet().isEmpty()
if (scan.getFamilyMap().keySet().isEmpty()
|| scan.getFamilyMap().size() > 1) {
// In a Scan operation where the family map is greater than 1 or equal to 0,
// we handle this by appending the column family to the qualifier on the client side.
Expand Down Expand Up @@ -1107,8 +1105,7 @@ public List<ResultScanner> call() throws IOException {
ObTableQuery obTableQuery;
ObHTableFilter filter;
try {
if (scan.getFamilyMap().keySet() == null
|| scan.getFamilyMap().keySet().isEmpty()
if (scan.getFamilyMap().keySet().isEmpty()
|| scan.getFamilyMap().size() > 1) {
// In a Scan operation where the family map is greater than 1 or equal to 0,
// we handle this by appending the column family to the qualifier on the client side.
Expand Down Expand Up @@ -1950,20 +1947,20 @@ public static ObTableBatchOperation buildObTableBatchOperation(List<Mutation> ro
Map<String, Integer> indexMap = new HashMap<>();
for (Mutation row : rowList) {
if (row instanceof Put) {
opType = OHOpType.INSERT_OR_UPDATE;
opType = OHOpType.Put;
} else if (row instanceof Delete) {
opType = OHOpType.Delete;
} else if (row instanceof Increment) {
opType = OHOpType.Increment;
} else if (row instanceof Append) {
opType = OHOpType.APPEND;
opType = OHOpType.Append;
} else {
throw new FeatureNotSupportedException("not supported other type");
}
Set<Map.Entry<byte[], List<Cell>>> familyCellMap = row.getFamilyCellMap().entrySet();

for (Map.Entry<byte[], List<Cell>> familyWithCells : familyCellMap) {
if (opType == OHOpType.Increment || opType == OHOpType.APPEND) {
if (opType == OHOpType.Increment || opType == OHOpType.Append) {
indexMap.clear();
for (int i = 0; i < familyWithCells.getValue().size(); i++) {
Cell cell = familyWithCells.getValue().get(i);
Expand Down Expand Up @@ -2002,12 +1999,12 @@ private com.alipay.oceanbase.rpc.mutation.Mutation buildMutation(Cell kv,
property = new Object[] { CellUtil.cloneValue(new_cell), TTL };
}
switch (operationType) {
case INSERT_OR_UPDATE:
case Put:
return com.alipay.oceanbase.rpc.mutation.Mutation.getInstance(INSERT_OR_UPDATE,
ROW_KEY_COLUMNS,
new Object[] { CellUtil.cloneRow(new_cell), CellUtil.cloneQualifier(new_cell),
new_cell.getTimestamp() }, property_columns, property);
case APPEND:
case Append:
return com.alipay.oceanbase.rpc.mutation.Mutation.getInstance(APPEND,
ROW_KEY_COLUMNS,
new Object[] { CellUtil.cloneRow(new_cell), CellUtil.cloneQualifier(new_cell),
Expand Down Expand Up @@ -2116,7 +2113,7 @@ private BatchOperation buildBatchOperation(String tableName, List<? extends Row>
List<Cell> keyValueList = entry.getValue();
for (Cell kv : keyValueList) {
singleOpResultNum++;
batch.addOperation(buildMutation(kv, OHOpType.INSERT_OR_UPDATE,
batch.addOperation(buildMutation(kv, OHOpType.Put,
isTableGroup, family, put.getTTL()));
}
}
Expand Down Expand Up @@ -2156,37 +2153,40 @@ public static ObTableOperation buildObTableOperation(Cell kv, OHOpType operation
property = new Object[] { CellUtil.cloneValue(kv), TTL };
}
switch (operationType) {
case INSERT_OR_UPDATE:
return getInstance(
INSERT_OR_UPDATE,
new Object[] { CellUtil.cloneRow(kv), CellUtil.cloneQualifier(kv),
kv.getTimestamp() }, property_columns, property);
case Put:
case Increment:
case Append:
ObTableOperationType type;
if (operationType == OHOpType.Put) {
type = INSERT_OR_UPDATE;
} else if (operationType == OHOpType.Increment) {
type = INCREMENT;
} else {
type = APPEND;
}
return getInstance(
INCREMENT,
new Object[] { CellUtil.cloneRow(kv), CellUtil.cloneQualifier(kv),
kv.getTimestamp() }, property_columns, property);
case APPEND:
return getInstance(
APPEND,
type,
new Object[] { CellUtil.cloneRow(kv), CellUtil.cloneQualifier(kv),
kv.getTimestamp() }, property_columns, property);
case Delete:
Cell.Type type = kv.getType();
if (type == Cell.Type.Delete) {
Cell.Type delType = kv.getType();
if (delType == Cell.Type.Delete) {
return getInstance(
DEL,
new Object[] { CellUtil.cloneRow(kv), CellUtil.cloneQualifier(kv),
kv.getTimestamp() }, null, null);
} else if (type == Cell.Type.DeleteColumn) {
} else if (delType == Cell.Type.DeleteColumn) {
return getInstance(
DEL,
new Object[] { CellUtil.cloneRow(kv), CellUtil.cloneQualifier(kv),
-kv.getTimestamp() }, null, null);
} else if (delType == Cell.Type.DeleteFamily) {
return getInstance(DEL,
new Object[] { CellUtil.cloneRow(kv), null, -kv.getTimestamp() }, null,
null);
} else {
throw new IllegalArgumentException("illegal delete type " + operationType);
}
case DeleteFamily:
return getInstance(DEL,
new Object[] { CellUtil.cloneRow(kv), null, -kv.getTimestamp() }, null, null);
default:
throw new IllegalArgumentException("illegal mutation type " + operationType);
}
Expand Down Expand Up @@ -2214,16 +2214,6 @@ private ObTableQueryAsyncRequest buildObTableQueryAsyncRequest(ObTableQuery obTa
return asyncRequest;
}

public static ObTableBatchOperationRequest buildObTableBatchOperationRequest(ObTableBatchOperation obTableBatchOperation,
String targetTableName) {
ObTableBatchOperationRequest request = new ObTableBatchOperationRequest();
request.setTableName(targetTableName);
request.setReturningAffectedRows(true);
request.setEntityType(ObTableEntityType.HKV);
request.setBatchOperation(obTableBatchOperation);
return request;
}

private ObTableQueryAndMutateRequest buildObTableQueryAndMutateRequest(ObTableQuery obTableQuery,
ObTableBatchOperation obTableBatchOperation,
String targetTableName) {
Expand Down Expand Up @@ -2310,7 +2300,7 @@ public Pair<byte[][], byte[][]> getStartEndKeys() throws IOException {
}

public static enum OHOpType {
INSERT_OR_UPDATE, APPEND, Delete, DeleteAll, DeleteColumn, DeleteFamily, DeleteFamilyVersion, Increment
Put, Append, Delete, DeleteAll, DeleteColumn, DeleteFamily, DeleteFamilyVersion, Increment
}

public static OHOpType getDeleteType(Cell.Type type) {
Expand Down