From b9178b6a434dd85855e27271bc9d720132f4fc85 Mon Sep 17 00:00:00 2001 From: "linguantian.lgt" Date: Tue, 18 Aug 2026 20:04:20 +0800 Subject: [PATCH] perf: port client CPU optimizations to hbase_2.0 --- pom.xml | 2 +- .../com/alipay/oceanbase/hbase/OHTable.java | 537 +++++++++++++----- .../hbase/result/ClientStreamScanner.java | 125 ++-- .../hbase/result/OHBaseResultCell.java | 214 +++++++ .../oceanbase/hbase/util/OHBaseFuncUtils.java | 73 ++- .../hbase/util/OHBufferedMutatorImpl.java | 3 +- .../hbase/OHTableBatchGetResultTest.java | 142 +++++ .../hbase/OHTableCompactPutCellTest.java | 89 +++ .../hbase/OHTableGetMaxRowResultTest.java | 400 +++++++++++++ .../hbase/OHTablePutSkipCellCloneTest.java | 161 ++++++ .../hbase/OHTablePutValidationTest.java | 128 +++++ .../ClientStreamScannerCompactResultTest.java | 119 ++++ .../hbase/result/OHBaseResultCellTest.java | 124 ++++ 13 files changed, 1895 insertions(+), 222 deletions(-) create mode 100644 src/main/java/com/alipay/oceanbase/hbase/result/OHBaseResultCell.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/OHTableBatchGetResultTest.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/OHTableCompactPutCellTest.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/OHTableGetMaxRowResultTest.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/OHTablePutSkipCellCloneTest.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/OHTablePutValidationTest.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/result/ClientStreamScannerCompactResultTest.java create mode 100644 src/test/java/com/alipay/oceanbase/hbase/result/OHBaseResultCellTest.java diff --git a/pom.xml b/pom.xml index 2a29f087..5787cce8 100644 --- a/pom.xml +++ b/pom.xml @@ -54,7 +54,7 @@ ${project.encoding} UTF-8 1.7.21 - 2.4.0 + 2.4.1-SNAPSHOT diff --git a/src/main/java/com/alipay/oceanbase/hbase/OHTable.java b/src/main/java/com/alipay/oceanbase/hbase/OHTable.java index c4d25fc5..d286eeb2 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/OHTable.java +++ b/src/main/java/com/alipay/oceanbase/hbase/OHTable.java @@ -24,9 +24,11 @@ import com.alipay.oceanbase.hbase.metrics.MetricsImporter; import com.alipay.oceanbase.hbase.metrics.OHMetrics; import com.alipay.oceanbase.hbase.result.ClientStreamScanner; +import com.alipay.oceanbase.hbase.result.OHBaseResultCell; import com.alipay.oceanbase.hbase.util.*; import com.alipay.oceanbase.rpc.ObGlobal; import com.alipay.oceanbase.rpc.ObTableClient; +import com.alipay.oceanbase.rpc.util.ObBytesString; import com.alipay.oceanbase.rpc.exception.ObTableException; import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException; import com.alipay.oceanbase.rpc.location.model.partition.ObPair; @@ -594,7 +596,7 @@ public boolean[] existsAll(List gets) throws IOException { @Override boolean[] execute() throws IOException { boolean[] ret = new boolean[gets.size()]; - List newGets = new ArrayList<>(); + List newGets = new ArrayList<>(gets.size()); // if just checkExistOnly, batch get will not return any result or row count // therefore we have to set checkExistOnly as false and so the result can be returned for (Get get : gets) { @@ -791,16 +793,20 @@ private void innerBatchImpl(final List actions, final Object[] re throw new AssertionError("results.length"); } } - BatchError batchError = new BatchError(); obTableClient.setRuntimeBatchExecutor(executePool); - List resultMapSingleOp = new LinkedList<>(); if (!ObGlobal.isHBaseBatchSupport()) { + BatchError batchError = new BatchError(); try { compatOldServerBatch(actions, results, batchError); } catch (Exception e) { throw new IOException(tableNameString + " table occurred unexpected error." , e); } - } else if (OHBaseFuncUtils.isAllPut(opType, actions) && OHBaseFuncUtils.isHBasePutPefSupport(obTableClient, enablePutOptimization)) { + if (batchError.hasErrors()) { + throw batchError.makeException(); + } + return; + } + if (OHBaseFuncUtils.isAllPut(opType, actions) && OHBaseFuncUtils.isHBasePutPefSupport(obTableClient, enablePutOptimization)) { // only support Put now ObHbaseRequest request = buildHbaseRequest(actions, opType); try { @@ -813,80 +819,147 @@ private void innerBatchImpl(final List actions, final Object[] re } catch (Exception e) { throw new IOException(tableNameString + " table occurred unexpected error." , e); } - } else { - String realTableName = getTargetTableName(actions); - BatchOperation batch = buildBatchOperation(realTableName, actions, - tableNameString.equals(realTableName), resultMapSingleOp); - batch.setHbaseOpType(opType); - BatchOperationResult tmpResults; - try { - tmpResults = batch.execute(); - } catch (Exception e) { - throw new IOException(tableNameString + " table occurred unexpected error.", e); + return; + } + BatchError batchError = new BatchError(); + boolean pureGetBatch = true; + for (Row action : actions) { + if (!(action instanceof Get)) { + pureGetBatch = false; + break; } - int index = 0; - for (int i = 0; i != actions.size(); ++i) { - if (tmpResults.getResults().get(index) instanceof ObTableException) { - if (results != null) { - results[i] = tmpResults.getResults().get(index); - } - batchError.add((ObTableException) tmpResults.getResults().get(index), actions.get(i), null); - } else if (actions.get(i) instanceof Get) { - if (results != null) { - // get results have been wrapped in MutationResult, need to fetch it - if (tmpResults.getResults().get(index) instanceof MutationResult) { - MutationResult mutationResult = (MutationResult) tmpResults.getResults().get(index); - ObPayload innerResult = mutationResult.getResult(); - if (innerResult instanceof ObTableSingleOpResult) { - ObTableSingleOpResult singleOpResult = (ObTableSingleOpResult) innerResult; - List cells = generateGetResult(singleOpResult); - results[i] = Result.create(cells); - } else { - throw new ObTableUnexpectedException("Unexpected type of result in MutationResult"); - } + } + List resultMapSingleOp = pureGetBatch ? null + : new ArrayList<>(actions.size()); + String realTableName = getTargetTableName(actions); + BatchOperation batch = buildBatchOperation(realTableName, actions, + tableNameString.equals(realTableName), resultMapSingleOp); + batch.setHbaseOpType(opType); + BatchOperationResult tmpResults; + try { + tmpResults = batch.execute(); + } catch (Exception e) { + throw new IOException(tableNameString + " table occurred unexpected error.", e); + } + List batchResults = tmpResults.getResults(); + if (pureGetBatch) { + consumePureGetBatchResults(actions, results, batchResults, batchError); + if (batchError.hasErrors()) { + throw batchError.makeException(); + } + return; + } + int index = 0; + for (int i = 0; i != actions.size(); ++i) { + if (batchResults.get(index) instanceof ObTableException) { + if (results != null) { + results[i] = batchResults.get(index); + } + batchError.add((ObTableException) batchResults.get(index), actions.get(i), null); + } else if (actions.get(i) instanceof Get) { + if (results != null) { + if (batchResults.get(index) instanceof MutationResult) { + MutationResult mutationResult = (MutationResult) batchResults.get(index); + ObPayload innerResult = mutationResult.getResult(); + if (innerResult instanceof ObTableSingleOpResult) { + ObTableSingleOpResult singleOpResult = (ObTableSingleOpResult) innerResult; + List cells = generateGetResult(singleOpResult); + results[i] = Result.create(cells); } else { - throw new ObTableUnexpectedException("Unexpected type of result in batch"); + throw new ObTableUnexpectedException("Unexpected type of result in MutationResult"); } - } - } else { - if (results != null) { - results[i] = new Result(); + } else { + throw new ObTableUnexpectedException("Unexpected type of result in batch"); } } - index += resultMapSingleOp.get(i); + } else { + if (results != null) { + results[i] = new Result(); + } } + index += resultMapSingleOp.get(i); } if (batchError.hasErrors()) { throw batchError.makeException(); } } - private List generateGetResult(ObTableSingleOpResult getResult) throws IOException { - List cells = new ArrayList<>(); + @VisibleForTesting + void consumePureGetBatchResults(List actions, Object[] results, + List batchResults, BatchError batchError) + throws IOException { + if (batchResults.isEmpty() && actions.size() == 1) { + if (results != null) { + results[0] = Result.create(Collections. emptyList()); + } + return; + } + if (batchResults.size() != actions.size()) { + throw new ObTableUnexpectedException( + "Unexpected pure Get batch result count, expected=" + actions.size() + ", actual=" + + batchResults.size()); + } + for (int i = 0; i < actions.size(); i++) { + Object batchResult = batchResults.get(i); + if (batchResult instanceof ObTableException) { + if (results != null) { + results[i] = batchResult; + } + batchError.add((ObTableException) batchResult, actions.get(i), null); + continue; + } + if (results == null) { + continue; + } + if (!(batchResult instanceof MutationResult)) { + throw new ObTableUnexpectedException("Unexpected type of result in pure Get batch"); + } + ObPayload innerResult = ((MutationResult) batchResult).getResult(); + if (!(innerResult instanceof ObTableSingleOpResult)) { + throw new ObTableUnexpectedException( + "Unexpected type of inner result in pure Get batch"); + } + results[i] = Result.create(generateGetResult((ObTableSingleOpResult) innerResult)); + } + } + + @VisibleForTesting + List generateGetResult(ObTableSingleOpResult getResult) throws IOException { ObTableSingleOpEntity singleOpEntity = getResult.getEntity(); - // all values queried by this get are contained in properties - // qualifier in batch get result is always appended after family - List propertiesValues = singleOpEntity.getPropertiesValues(); - int valueIdx = 0; - while (valueIdx < propertiesValues.size()) { - // values in propertiesValues like: [ K, Q, T, V, K, Q, T, V ... ] - // we need to retrieve K Q T V and construct them to cells: [ cell_0, cell_1, ... ] - byte[][] familyAndQualifier = new byte[2][]; + ObHBaseCellBatch compactBatch = singleOpEntity.getHBaseCellBatch(); + if (compactBatch != null) { + List cells = new ArrayList<>(compactBatch.size()); try { - // split family and qualifier - familyAndQualifier = OHBaseFuncUtils - .extractFamilyFromQualifier((byte[]) propertiesValues.get(valueIdx + 1).getValue()); + for (int cellIndex = 0; cellIndex < compactBatch.size(); cellIndex++) { + addResultCell(cells, compactBatch.getRowKey(cellIndex), + compactBatch.getQualifier(cellIndex), compactBatch.getTimestamp(cellIndex), + compactBatch.getValue(cellIndex), true, HConstants.EMPTY_BYTE_ARRAY); + } } catch (Exception e) { throw new IOException(e); } - KeyValue kv = new KeyValue((byte[]) propertiesValues.get(valueIdx).getValue(),//K - familyAndQualifier[0], // family - familyAndQualifier[1], // qualifiermat - (Long) propertiesValues.get(valueIdx + 2).getValue(), // T - (byte[]) propertiesValues.get(valueIdx + 3).getValue()// V - ); - cells.add(kv); - valueIdx += 4; + return cells; + } + // all values queried by this get are contained in properties + // qualifier in batch get result is always appended after family + List propertiesValues = singleOpEntity.getPropertiesValues(); + int propertyCount = propertiesValues.size(); + if ((propertyCount & 3) != 0) { + throw new IOException("Malformed Batch Get K/Q/T/V result, property count=" + + propertyCount); + } + List cells = new ArrayList<>(propertyCount / 4); + try { + for (int valueIdx = 0; valueIdx < propertyCount; valueIdx += 4) { + byte[] rowKey = (byte[]) propertiesValues.get(valueIdx).getValue(); + byte[] familyQualifier = (byte[]) propertiesValues.get(valueIdx + 1).getValue(); + long timestamp = (Long) propertiesValues.get(valueIdx + 2).getValue(); + byte[] value = (byte[]) propertiesValues.get(valueIdx + 3).getValue(); + addResultCell(cells, rowKey, familyQualifier, timestamp, value, true, + HConstants.EMPTY_BYTE_ARRAY); + } + } catch (Exception e) { + throw new IOException(e); } return cells; } @@ -957,41 +1030,135 @@ public static int compareByteArray(byte[] bt1, byte[] bt2) { return bt1.length - bt2.length; } - private void getMaxRowFromResult(AbstractQueryStreamResult clientQueryStreamResult, - List keyValueList, boolean isTableGroup, byte[] family) - throws Exception { - byte[][] familyAndQualifier = new byte[2][]; - KeyValue kv = null; + private void addRowToResultCells(List cells, List row, byte[] rowKey, + boolean isTableGroup, byte[] family) throws Exception { + byte[] qualifier = (byte[]) row.get(1).getValue(); + long timestamp = (Long) row.get(2).getValue(); + byte[] value = (byte[]) row.get(3).getValue(); + addResultCell(cells, rowKey, qualifier, timestamp, value, isTableGroup, family); + } + + private void addResultCell(List cells, byte[] rowKey, byte[] qualifier, long timestamp, + byte[] value, boolean isTableGroup, byte[] family) throws Exception { + if (isTableGroup) { + cells.add(OHBaseResultCell.createTableGroup(rowKey, qualifier, timestamp, value)); + } else { + cells.add(OHBaseResultCell.create(rowKey, family, qualifier, timestamp, value)); + } + } + + /** + * A point Get uses an exact row range, so all returned cells must belong to the requested row. + * Validate the invariant once and reuse the returned rowkey while assembling result cells. + */ + private boolean fillPointGetFromResult(AbstractQueryStreamResult clientQueryStreamResult, + List cells, boolean isTableGroup, byte[] family, + byte[] expectedRowKey, boolean checkExistenceOnly) + throws Exception { + byte[] canonicalRowKey = null; while (clientQueryStreamResult.next()) { - List row = clientQueryStreamResult.getRow(); - if (row.isEmpty()) { - // Currently, checkExistOnly is set, and if the row exists, it returns an empty row. - keyValueList.add(new KeyValue()); - return; + if (checkExistenceOnly) { + // An existence-only response contains an empty row when the requested row exists. + return true; + } + + byte[] actualRowKey; + byte[] qualifier; + long timestamp; + byte[] value; + if (clientQueryStreamResult.isCurrentHBaseCell()) { + ObHBaseCellBatch batch = clientQueryStreamResult.getCurrentHBaseCellBatch(); + int index = clientQueryStreamResult.getCurrentHBaseCellIndex(); + actualRowKey = batch.getRowKey(index); + qualifier = batch.getQualifier(index); + timestamp = batch.getTimestamp(index); + value = batch.getValue(index); } else { - if (kv == null - || compareByteArray(CellUtil.cloneRow(kv), (byte[]) row.get(0).getValue()) <= 0) { - if (kv != null - && compareByteArray(CellUtil.cloneRow(kv), (byte[]) row.get(0).getValue()) != 0) { - keyValueList.clear(); - } - if (isTableGroup) { - // split family and qualifier - familyAndQualifier = OHBaseFuncUtils - .extractFamilyFromQualifier((byte[]) row.get(1).getValue()); - } else { - familyAndQualifier[0] = family; - familyAndQualifier[1] = (byte[]) row.get(1).getValue(); - } - kv = new KeyValue((byte[]) row.get(0).getValue(),//K - familyAndQualifier[0], // family - familyAndQualifier[1], // qualifiermat - (Long) row.get(2).getValue(), // T - (byte[]) row.get(3).getValue() // V - ); - keyValueList.add(kv); + List row = clientQueryStreamResult.getRow(); + actualRowKey = (byte[]) row.get(0).getValue(); + qualifier = (byte[]) row.get(1).getValue(); + timestamp = (Long) row.get(2).getValue(); + value = (byte[]) row.get(3).getValue(); + } + if (canonicalRowKey == null) { + if (!Bytes.equals(expectedRowKey, actualRowKey)) { + throw new ObTableUnexpectedException( + "point Get returned an unexpected rowkey, expected length=" + + expectedRowKey.length + ", actual length=" + actualRowKey.length); } + canonicalRowKey = actualRowKey; + } + addResultCell(cells, canonicalRowKey, qualifier, timestamp, value, isTableGroup, family); + } + return canonicalRowKey != null; + } + + /** + * closestRowBefore may collect candidates from multiple partitions. Keep only cells belonging + * to the greatest returned rowkey without cloning the rowkey from a temporary result cell. + */ + private boolean getMaxRowFromResult(AbstractQueryStreamResult clientQueryStreamResult, + List cells, boolean isTableGroup, byte[] family, + boolean checkExistenceOnly) throws Exception { + byte[] currentMaxRowKey = null; + while (clientQueryStreamResult.next()) { + if (checkExistenceOnly) { + // An existence-only response contains an empty row when a candidate row exists. + return true; + } + + byte[] rowKey; + byte[] qualifier; + long timestamp; + byte[] value; + if (clientQueryStreamResult.isCurrentHBaseCell()) { + ObHBaseCellBatch batch = clientQueryStreamResult.getCurrentHBaseCellBatch(); + int index = clientQueryStreamResult.getCurrentHBaseCellIndex(); + rowKey = batch.getRowKey(index); + qualifier = batch.getQualifier(index); + timestamp = batch.getTimestamp(index); + value = batch.getValue(index); + } else { + List row = clientQueryStreamResult.getRow(); + rowKey = (byte[]) row.get(0).getValue(); + qualifier = (byte[]) row.get(1).getValue(); + timestamp = (Long) row.get(2).getValue(); + value = (byte[]) row.get(3).getValue(); + } + if (currentMaxRowKey != null) { + int rowComparison = compareByteArray(currentMaxRowKey, rowKey); + if (rowComparison > 0) { + continue; + } + if (rowComparison < 0) { + cells.clear(); + currentMaxRowKey = rowKey; + } + } else { + currentMaxRowKey = rowKey; + } + addResultCell(cells, rowKey, qualifier, timestamp, value, isTableGroup, family); + } + return currentMaxRowKey != null; + } + + private Result createGetResult(List cells) { + return Result.create(cells); + } + + private void addQueryResultToKeyValueList(ObTableQueryResult queryResult, List keyValues, + byte[] family) throws Exception { + ObHBaseCellBatch batch = queryResult.getHBaseCellBatch(); + if (batch != null) { + for (int i = 0; i < batch.size(); i++) { + keyValues.add(new KeyValue(batch.getRowKey(i), family, batch.getQualifier(i), batch + .getTimestamp(i), batch.getValue(i))); } + return; + } + for (List row : queryResult.getPropertiesRows()) { + keyValues.add(new KeyValue((byte[]) row.get(0).getValue(), family, (byte[]) row.get(1) + .getValue(), (Long) row.get(2).getValue(), (byte[]) row.get(3).getValue())); } } @@ -1052,8 +1219,9 @@ private Result innerGetImpl(final Get get, OHOperationType opType) throws IOExce ServerCallable serverCallable = new ServerCallable(configuration, obTableClient, tableNameString, get.getRow(), get.getRow(), operationTimeout) { public Result call() throws IOException { - List keyValueList = new ArrayList<>(); + List cells = new ArrayList<>(); byte[] family = new byte[] {}; + boolean exists = false; ObTableQuery obTableQuery; try { if (get.getFamilyMap().keySet().isEmpty() @@ -1072,7 +1240,13 @@ public Result call() throws IOException { ObTableClientQueryAsyncStreamResult clientQueryStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient .execute(request); - getMaxRowFromResult(clientQueryStreamResult, keyValueList, true, family); + if (get.isClosestRowBefore()) { + exists = getMaxRowFromResult(clientQueryStreamResult, cells, true, + family, get.isCheckExistenceOnly()); + } else { + exists = fillPointGetFromResult(clientQueryStreamResult, cells, + true, family, get.getRow(), get.isCheckExistenceOnly()); + } } else { for (Map.Entry> entry : get.getFamilyMap() .entrySet()) { @@ -1094,8 +1268,14 @@ public Result call() throws IOException { configuration), isWeakRead(get), opType); ObTableClientQueryStreamResult clientQueryStreamResult = (ObTableClientQueryStreamResult) obTableClient .execute(request); - getMaxRowFromResult(clientQueryStreamResult, keyValueList, false, - family); + if (get.isClosestRowBefore()) { + exists = getMaxRowFromResult(clientQueryStreamResult, cells, false, + family, get.isCheckExistenceOnly()); + } else { + exists = fillPointGetFromResult(clientQueryStreamResult, cells, + false, family, get.getRow(), + get.isCheckExistenceOnly()); + } } } } catch (Exception e) { @@ -1103,11 +1283,11 @@ public Result call() throws IOException { + Bytes.toString(family) + " error.", e); } if (get.isCheckExistenceOnly()) { - return Result.create(null, !keyValueList.isEmpty()); + return Result.create(null, exists); } - // sort keyValues - OHBaseFuncUtils.sortHBaseResult(keyValueList); - return Result.create(keyValueList); + // sort result cells + OHBaseFuncUtils.sortHBaseResult(cells); + return createGetResult(cells); } }; return executeServerCallable(serverCallable); @@ -1123,8 +1303,19 @@ Result[] execute() throws IOException { if (ObGlobal.isHBaseBatchGetSupport()) { // get only supported in BatchSupport version innerBatchImpl(gets, results, opType); } else { + List> futures = new ArrayList<>(gets.size()); + for (int i = 0; i < gets.size(); i++) { + int index = i; + Future future = executePool.submit(() -> innerGetImpl( + gets.get(index), opType)); + futures.add(future); + } for (int i = 0; i < gets.size(); i++) { - results[i] = innerGetImpl(gets.get(i), opType); // still use list type even executing gets one by one in loop + try { + results[i] = futures.get(i).get(); + } catch (Exception e) { + throw new RuntimeException("gets occur error. index:{" + i + "}", e); + } } } return results; @@ -1360,9 +1551,11 @@ public Void execute() throws IOException { } private void doPut(List puts, OHOperationType opType) throws IOException { + if (puts.isEmpty()) { + return; + } for (Put put : puts) { - validatePut(put); - checkFamilyViolation(put.getFamilyCellMap().keySet(), true); + validatePutMutation(put); } if (OHBaseFuncUtils.isHBasePutPefSupport(obTableClient, enablePutOptimization)) { flushCommitsV2(puts, opType); @@ -1371,23 +1564,38 @@ private void doPut(List puts, OHOperationType opType) throws IOException { } } + /** + * Put validation shared by direct Table puts and BufferedMutator. + */ + public void validatePutMutation(Put put) { + NavigableMap> familyCellMap = put.getFamilyCellMap(); + validatePut(put, familyCellMap, maxKeyValueSize); + checkFamilyViolation(familyCellMap.keySet(), true); + } + /** * 校验 put 里的参数是否合法，需要传入 family ，并且 keyvalue 的 size 不能太大 * @param put the put */ - private void validatePut(Put put) { + public static void validatePut(Put put, int maxKeyValueSize) { + validatePut(put, put.getFamilyCellMap(), maxKeyValueSize); + } + + private static void validatePut(Put put, NavigableMap> familyCellMap, + int maxKeyValueSize) { if (put.isEmpty()) { throw new IllegalArgumentException("No columns to insert"); } if (maxKeyValueSize > 0) { - for (Map.Entry> entry : put.getFamilyCellMap().entrySet()) { + for (Map.Entry> entry : familyCellMap.entrySet()) { if (entry.getKey() == null || entry.getKey().length == 0) { throw new IllegalArgumentException("family is empty"); } - for (Cell kv : entry.getValue()) { - if (kv.getRowLength() + kv.getValueLength() + kv.getQualifierLength() - + Bytes.toBytes(kv.getTimestamp()).length + kv.getFamilyLength() > maxKeyValueSize) { + for (Cell cell : entry.getValue()) { + int cellLength = cell instanceof KeyValue ? ((KeyValue) cell).getLength() + : KeyValueUtil.length(cell); + if (cellLength > maxKeyValueSize) { throw new IllegalArgumentException("KeyValue size too large"); } } @@ -1627,15 +1835,7 @@ Result execute() throws IOException { } ObTableQueryResult queryResult = result.getAffectedEntity(); List keyValues = new ArrayList(); - for (List row : queryResult.getPropertiesRows()) { - byte[] k = (byte[]) row.get(0).getValue(); - byte[] q = (byte[]) row.get(1).getValue(); - long t = (Long) row.get(2).getValue(); - byte[] v = (byte[]) row.get(3).getValue(); - KeyValue kv = new KeyValue(k, f, q, t, v); - - keyValues.add(kv); - } + addQueryResultToKeyValueList(queryResult, keyValues, f); return Result.create(keyValues); } catch (Exception e) { throw new IOException("append table " + tableNameString + " error.", e); @@ -1684,14 +1884,7 @@ Result execute() throws IOException { } ObTableQueryResult queryResult = result.getAffectedEntity(); List keyValues = new ArrayList(); - for (List row : queryResult.getPropertiesRows()) { - byte[] k = (byte[]) row.get(0).getValue(); - byte[] q = (byte[]) row.get(1).getValue(); - long t = (Long) row.get(2).getValue(); - byte[] v = (byte[]) row.get(3).getValue(); - KeyValue kv = new KeyValue(k, f, q, t, v); - keyValues.add(kv); - } + addQueryResultToKeyValueList(queryResult, keyValues, f); return Result.create(keyValues); } catch (Exception e) { throw new IOException("increment table " + tableNameString + " error.", e); @@ -2330,10 +2523,11 @@ private com.alipay.oceanbase.rpc.mutation.Mutation buildMutation(Cell kv, switch (kvType) { case Put: String[] propertyColumns = V_COLUMNS; - Object[] property = new Object[] { CellUtil.cloneValue(newCell) }; + Object value = CellUtil.cloneValue(newCell); + Object[] property = new Object[] { value }; if (TTL != Long.MAX_VALUE) { propertyColumns = PROPERTY_COLUMNS; - property = new Object[] { CellUtil.cloneValue(newCell), TTL }; + property = new Object[] { value, TTL }; } return com.alipay.oceanbase.rpc.mutation.Mutation.getInstance(operationType, ROW_KEY_COLUMNS, @@ -2516,7 +2710,9 @@ private BatchOperation buildBatchOperation(String tableName, List throw new FeatureNotSupportedException( "not supported other type in batch yet,only support get, put and delete"); } - resultMapSingleOp.add(singleOpResultNum); + if (resultMapSingleOp != null) { + resultMapSingleOp.add(singleOpResultNum); + } } // only set weak read consistency when all operations are Get and all Get operations are weak read if (getOperationNum == actions.size() && allGetIsWeakRead) { @@ -2528,13 +2724,14 @@ private BatchOperation buildBatchOperation(String tableName, List return batch; } - private ObHbaseRequest buildHbaseRequest(List actions, OHOperationType hbaseOpType) + @VisibleForTesting + ObHbaseRequest buildHbaseRequest(List actions, OHOperationType hbaseOpType) throws FeatureNotSupportedException, IllegalArgumentException, IOException { ObHbaseRequest request = new ObHbaseRequest(); ObTableOperationType opType = null; - List keys = new ArrayList<>(); + List keys = new ArrayList<>(actions.size()); List cfRowsArray = new ArrayList<>(); Map cfRowsMap = new HashMap<>(); int keyIndex = 0; @@ -2545,35 +2742,28 @@ private ObHbaseRequest buildHbaseRequest(List actions, OHOperatio if (put.isEmpty()) { throw new IllegalArgumentException("No columns to put for item"); } - boolean isCellTTL = false; long ttl = put.getTTL(); - if (ttl != Long.MAX_VALUE) { - isCellTTL = true; - } - keys.add(ObObj.getInstance(put.getRow())); + keys.add(ObObj.hbasePutVarchar(put.getRow())); + // HBase 2.0 builds and executes this request synchronously, so the encoder can + // reference Cell Q/V regions until they are copied into the RPC buffer. + boolean shareCellBytes = true; for (Map.Entry> entry : put.getFamilyCellMap().entrySet()) { String family = Bytes.toString(entry.getKey()); ObHbaseCfRows sameCfRows = cfRowsMap.get(family); if (sameCfRows == null) { sameCfRows = new ObHbaseCfRows(); + sameCfRows.reserveKeyRuns(actions.size()); String realTableName = getTargetTableName(tableNameString, family, configuration); sameCfRows.setRealTableName(realTableName); cfRowsMap.put(family, sameCfRows); cfRowsArray.add(sameCfRows); } List keyValueList = entry.getValue(); - List cells = new ArrayList<>(); + sameCfRows.reserveAdditionalCompactCells(keyValueList.size()); + sameCfRows.beginCompactKeyCells(keyIndex, keyValueList.size(), ttl); for (Cell kv : keyValueList) { - ObHbaseCell cell = new ObHbaseCell(isCellTTL); - cell.setQ(ObObj.getInstance(CellUtil.cloneQualifier(kv))); - cell.setT(ObObj.getInstance(-getEffectiveTimestampForWrite(kv.getTimestamp()))); // set timestamp as negative - cell.setV(ObObj.getInstance(CellUtil.cloneValue(kv))); - if (isCellTTL) { - cell.setTTL(ObObj.getInstance(ttl)); - } - cells.add(cell); + appendCompactPutCell(sameCfRows, kv, shareCellBytes); } - sameCfRows.add(keyIndex, cells.size(), cells); } } else { throw new FeatureNotSupportedException( @@ -2590,15 +2780,66 @@ private ObHbaseRequest buildHbaseRequest(List actions, OHOperatio return request; } - private ObTableOperation buildObTableOperation(Cell kv, + /** + * Put V2 Q/V payload for ObObj. When {@code share} is true, reference the Cell + * region without clone: contiguous arrays are returned as {@code byte[]}; sliced + * regions use an {@link ObBytesString} view. Encode copies once into the RPC buffer. + * HBase 2.0 callers may share only while building and executing the request synchronously; + * callers that retain the payload beyond the current call must pass {@code share=false}. + */ + @VisibleForTesting + static Object bytesForPutCell(Cell cell, boolean share, boolean qualifier) { + if (share) { + byte[] array = qualifier ? cell.getQualifierArray() : cell.getValueArray(); + int offset = qualifier ? cell.getQualifierOffset() : cell.getValueOffset(); + int length = qualifier ? cell.getQualifierLength() : cell.getValueLength(); + if (array == null) { + return new byte[0]; + } + if (offset == 0 && length == array.length) { + return array; + } + return new ObBytesString(array, offset, length); + } + return qualifier ? CellUtil.cloneQualifier(cell) : CellUtil.cloneValue(cell); + } + + private void appendCompactPutCell(ObHbaseCfRows cfRows, Cell cell, boolean shareCellBytes) { + byte[] qualifier; + int qualifierOffset; + int qualifierLength; + byte[] value; + int valueOffset; + int valueLength; + if (shareCellBytes) { + qualifier = cell.getQualifierArray(); + qualifierOffset = cell.getQualifierOffset(); + qualifierLength = cell.getQualifierLength(); + value = cell.getValueArray(); + valueOffset = cell.getValueOffset(); + valueLength = cell.getValueLength(); + } else { + qualifier = CellUtil.cloneQualifier(cell); + qualifierOffset = 0; + qualifierLength = qualifier.length; + value = CellUtil.cloneValue(cell); + valueOffset = 0; + valueLength = value.length; + } + cfRows.appendCompactCell(qualifier, qualifierOffset, qualifierLength, + -getEffectiveTimestampForWrite(cell.getTimestamp()), value, valueOffset, valueLength); + } + + ObTableOperation buildObTableOperation(Cell kv, ObTableOperationType operationType, Long TTL) { Cell.Type kvType = kv.getType(); String[] propertyColumns = V_COLUMNS; - Object[] property = new Object[] { CellUtil.cloneValue(kv) }; + Object value = CellUtil.cloneValue(kv); + Object[] property = new Object[] { value }; if (TTL != Long.MAX_VALUE) { propertyColumns = PROPERTY_COLUMNS; - property = new Object[] { CellUtil.cloneValue(kv), TTL }; + property = new Object[] { value, TTL }; } switch (kvType) { case Put: diff --git a/src/main/java/com/alipay/oceanbase/hbase/result/ClientStreamScanner.java b/src/main/java/com/alipay/oceanbase/hbase/result/ClientStreamScanner.java index 0b691881..6b9a0f4d 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/result/ClientStreamScanner.java +++ b/src/main/java/com/alipay/oceanbase/hbase/result/ClientStreamScanner.java @@ -25,6 +25,8 @@ import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj; import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.OHOperationType; import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.AbstractQueryStreamResult; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellBatch; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellRow; import com.alipay.oceanbase.rpc.stream.ObTableClientQueryAsyncStreamResult; import com.alipay.oceanbase.rpc.stream.ObTableClientQueryStreamResult; import org.apache.hadoop.classification.InterfaceAudience; @@ -89,55 +91,13 @@ public Result next() throws IOException { return null; } checkStatus(); - List startRow; - if (streamResult.next()) { - startRow = streamResult.getRow(); - } else { + if (!streamResult.next()) { return null; } - - byte[][] familyAndQualifier = new byte[2][]; - if (this.isTableGroup) { - // split family and qualifier - familyAndQualifier = OHBaseFuncUtils.extractFamilyFromQualifier((byte[]) startRow - .get(1).getValue()); - this.family = familyAndQualifier[0]; - } else { - familyAndQualifier[1] = (byte[]) startRow.get(1).getValue(); + if (streamResult.isCurrentHBaseCell()) { + return buildCompactResult(streamResult.drainCurrentHBaseRow()); } - - byte[] sk = (byte[]) startRow.get(0).getValue(); - byte[] sq = familyAndQualifier[1]; - long st = (Long) startRow.get(2).getValue(); - byte[] sv = (byte[]) startRow.get(3).getValue(); - KeyValue startKeyValue = new KeyValue(sk, family, sq, st, sv); - List keyValues = new ArrayList(); - keyValues.add(startKeyValue); - while (!streamResult.getCacheRows().isEmpty() && streamResult.next()) { - List row = streamResult.getRow(); - if (this.isTableGroup) { - // split family and qualifier - familyAndQualifier = OHBaseFuncUtils.extractFamilyFromQualifier((byte[]) row - .get(1).getValue()); - this.family = familyAndQualifier[0]; - } else { - familyAndQualifier[1] = (byte[]) row.get(1).getValue(); - } - byte[] k = (byte[]) row.get(0).getValue(); - byte[] q = familyAndQualifier[1]; - long t = (Long) row.get(2).getValue(); - byte[] v = (byte[]) row.get(3).getValue(); - if (Arrays.equals(sk, k)) { - // when rowKey is equal to the previous rowKey ,merge the result into the same result - keyValues.add(new KeyValue(k, family, q, t, v)); - } else { - streamResult.getCacheRows().addFirst(row); - break; - } - } - // sort keyValues - OHBaseFuncUtils.sortHBaseResult(keyValues); - return Result.create(keyValues); + return buildLegacyResult(streamResult.getRow()); } catch (Exception e) { if (importer != null) { importer.setIsFailedOp(true); @@ -160,6 +120,79 @@ public Result next() throws IOException { } } + private Result buildCompactResult(ObHBaseCellRow hbaseRow) { + List cells = new ArrayList(hbaseRow.getCellCount()); + byte[] rowKey = hbaseRow.getRowKey(); + for (int sliceIndex = 0; sliceIndex < hbaseRow.getSliceCount(); sliceIndex++) { + ObHBaseCellBatch batch = hbaseRow.getBatch(sliceIndex); + int toIndex = hbaseRow.getToIndex(sliceIndex); + for (int index = hbaseRow.getFromIndex(sliceIndex); index < toIndex; index++) { + addCompactResultCell(cells, rowKey, batch.getQualifier(index), + batch.getTimestamp(index), batch.getValue(index)); + } + } + OHBaseFuncUtils.sortHBaseResult(cells); + return createCompactResult(cells); + } + + private void addCompactResultCell(List cells, byte[] rowKey, byte[] qualifier, + long timestamp, byte[] value) { + if (isTableGroup) { + cells + .add(OHBaseFuncUtils.createTableGroupKeyValue(rowKey, qualifier, timestamp, value)); + } else { + cells.add(new KeyValue(rowKey, family, qualifier, timestamp, value)); + } + } + + private Result createCompactResult(List cells) { + return Result.create(cells); + } + + private Result buildLegacyResult(List startRow) throws Exception { + byte[][] familyAndQualifier = new byte[2][]; + if (this.isTableGroup) { + // split family and qualifier + familyAndQualifier = OHBaseFuncUtils.extractFamilyFromQualifier((byte[]) startRow + .get(1).getValue()); + this.family = familyAndQualifier[0]; + } else { + familyAndQualifier[1] = (byte[]) startRow.get(1).getValue(); + } + + byte[] sk = (byte[]) startRow.get(0).getValue(); + byte[] sq = familyAndQualifier[1]; + long st = (Long) startRow.get(2).getValue(); + byte[] sv = (byte[]) startRow.get(3).getValue(); + KeyValue startKeyValue = new KeyValue(sk, family, sq, st, sv); + List keyValues = new ArrayList(); + keyValues.add(startKeyValue); + while (!streamResult.getCacheRows().isEmpty() && streamResult.next()) { + List row = streamResult.getRow(); + if (this.isTableGroup) { + // split family and qualifier + familyAndQualifier = OHBaseFuncUtils.extractFamilyFromQualifier((byte[]) row + .get(1).getValue()); + this.family = familyAndQualifier[0]; + } else { + familyAndQualifier[1] = (byte[]) row.get(1).getValue(); + } + byte[] k = (byte[]) row.get(0).getValue(); + byte[] q = familyAndQualifier[1]; + long t = (Long) row.get(2).getValue(); + byte[] v = (byte[]) row.get(3).getValue(); + if (Arrays.equals(sk, k)) { + // when rowKey is equal to the previous rowKey ,merge the result into the same result + keyValues.add(new KeyValue(k, family, q, t, v)); + } else { + streamResult.getCacheRows().addFirst(row); + break; + } + } + OHBaseFuncUtils.sortHBaseResult(keyValues); + return Result.create(keyValues); + } + @Override public Result[] next(int nbRows) throws IOException { ArrayList resultSets = new ArrayList(nbRows); diff --git a/src/main/java/com/alipay/oceanbase/hbase/result/OHBaseResultCell.java b/src/main/java/com/alipay/oceanbase/hbase/result/OHBaseResultCell.java new file mode 100644 index 00000000..03467093 --- /dev/null +++ b/src/main/java/com/alipay/oceanbase/hbase/result/OHBaseResultCell.java @@ -0,0 +1,214 @@ +/*- + * #%L + * OBKV HBase Client Framework + * %% + * Copyright (C) 2022 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase.result; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.util.Bytes; + +import java.util.Objects; + +/** + * Immutable read-result cell backed by the decoded HBase field arrays. + */ +@InterfaceAudience.Private +public final class OHBaseResultCell implements Cell { + + private static final byte PUT_TYPE = KeyValue.Type.Put.getCode(); + private static final byte[] EMPTY = HConstants.EMPTY_BYTE_ARRAY; + + private final byte[] row; + private final byte[] familyArray; + private final int familyOffset; + private final int familyLength; + private final byte[] qualifierArray; + private final int qualifierOffset; + private final int qualifierLength; + private final long timestamp; + private final byte[] value; + + public static OHBaseResultCell create(byte[] row, byte[] family, byte[] qualifier, + long timestamp, byte[] value) { + return new OHBaseResultCell(row, family, 0, length(family), qualifier, 0, + length(qualifier), timestamp, value); + } + + public static OHBaseResultCell createTableGroup(byte[] row, byte[] familyQualifier, + long timestamp, byte[] value) { + Objects.requireNonNull(familyQualifier, "familyQualifier is null"); + int familyLength = findFamilyDelimiter(familyQualifier); + int qualifierOffset = familyLength + 1; + return new OHBaseResultCell(row, familyQualifier, 0, familyLength, familyQualifier, + qualifierOffset, familyQualifier.length - qualifierOffset, timestamp, value); + } + + private OHBaseResultCell(byte[] row, byte[] familyArray, int familyOffset, int familyLength, + byte[] qualifierArray, int qualifierOffset, int qualifierLength, + long timestamp, byte[] value) { + this.row = Objects.requireNonNull(row, "row is null"); + this.familyArray = Objects.requireNonNull(familyArray, "family is null"); + this.qualifierArray = Objects.requireNonNull(qualifierArray, "qualifier is null"); + this.value = value == null ? EMPTY : value; + checkRange(familyArray, familyOffset, familyLength, "family"); + checkRange(qualifierArray, qualifierOffset, qualifierLength, "qualifier"); + if (row.length > Short.MAX_VALUE) { + throw new IllegalArgumentException("row length " + row.length + " exceeds " + + Short.MAX_VALUE); + } + if (familyLength > Byte.MAX_VALUE) { + throw new IllegalArgumentException("family length " + familyLength + " exceeds " + + Byte.MAX_VALUE); + } + this.familyOffset = familyOffset; + this.familyLength = familyLength; + this.qualifierOffset = qualifierOffset; + this.qualifierLength = qualifierLength; + this.timestamp = timestamp; + } + + private static int length(byte[] value) { + return value == null ? 0 : value.length; + } + + private static void checkRange(byte[] array, int offset, int length, String field) { + if (offset < 0 || length < 0 || offset > array.length - length) { + throw new IndexOutOfBoundsException(field + " range is out of bounds"); + } + } + + private static int findFamilyDelimiter(byte[] familyQualifier) { + for (int i = 0; i < familyQualifier.length; i++) { + if (familyQualifier[i] == '\0') { + return i; + } + } + throw new RuntimeException("Cannot get family name"); + } + + @Override + public byte[] getRowArray() { + return row; + } + + @Override + public int getRowOffset() { + return 0; + } + + @Override + public short getRowLength() { + return (short) row.length; + } + + @Override + public byte[] getFamilyArray() { + return familyArray; + } + + @Override + public int getFamilyOffset() { + return familyOffset; + } + + @Override + public byte getFamilyLength() { + return (byte) familyLength; + } + + @Override + public byte[] getQualifierArray() { + return qualifierArray; + } + + @Override + public int getQualifierOffset() { + return qualifierOffset; + } + + @Override + public int getQualifierLength() { + return qualifierLength; + } + + @Override + public long getTimestamp() { + return timestamp; + } + + @Override + public byte getTypeByte() { + return PUT_TYPE; + } + + public long getMvccVersion() { + return 0L; + } + + @Override + public long getSequenceId() { + return 0L; + } + + @Override + public byte[] getValueArray() { + return value; + } + + @Override + public int getValueOffset() { + return 0; + } + + @Override + public int getValueLength() { + return value.length; + } + + @Override + public byte[] getTagsArray() { + return EMPTY; + } + + @Override + public int getTagsOffset() { + return 0; + } + + @Override + public int getTagsLength() { + return 0; + } + + public byte[] getValue() { + return Bytes.copy(value, 0, value.length); + } + + public byte[] getFamily() { + return Bytes.copy(familyArray, familyOffset, familyLength); + } + + public byte[] getQualifier() { + return Bytes.copy(qualifierArray, qualifierOffset, qualifierLength); + } + + public byte[] getRow() { + return Bytes.copy(row, 0, row.length); + } +} diff --git a/src/main/java/com/alipay/oceanbase/hbase/util/OHBaseFuncUtils.java b/src/main/java/com/alipay/oceanbase/hbase/util/OHBaseFuncUtils.java index eee513db..1b2ef0ff 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/util/OHBaseFuncUtils.java +++ b/src/main/java/com/alipay/oceanbase/hbase/util/OHBaseFuncUtils.java @@ -22,6 +22,7 @@ import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.OHOperationType; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Row; import org.apache.hadoop.hbase.util.Bytes; @@ -32,6 +33,51 @@ @InterfaceAudience.Private public class OHBaseFuncUtils { + private static final Comparator HBASE_CELL_COMPARATOR = new Comparator() { + @Override + public int compare(Cell cell1, Cell cell2) { + int familyComparison = Bytes.compareTo(cell1.getFamilyArray(), cell1.getFamilyOffset(), + cell1.getFamilyLength(), cell2.getFamilyArray(), cell2.getFamilyOffset(), + cell2.getFamilyLength()); + if (familyComparison != 0) { + return familyComparison; + } + + int qualifierComparison = Bytes.compareTo(cell1.getQualifierArray(), + cell1.getQualifierOffset(), cell1.getQualifierLength(), cell2.getQualifierArray(), + cell2.getQualifierOffset(), cell2.getQualifierLength()); + if (qualifierComparison != 0) { + return qualifierComparison; + } + + return Long.compare(cell2.getTimestamp(), cell1.getTimestamp()); + } + }; + + /** + * Build a TableGroup KeyValue directly from the protocol's {@code family\0qualifier} column. + * The offset constructor copies both ranges into the final KeyValue backing array and avoids + * allocating temporary family and qualifier arrays. + */ + public static KeyValue createTableGroupKeyValue(byte[] row, byte[] familyQualifier, + long timestamp, byte[] value) { + int familyLength = findFamilyDelimiter(familyQualifier); + int qualifierOffset = familyLength + 1; + return new KeyValue(row, 0, row == null ? 0 : row.length, familyQualifier, 0, familyLength, + familyQualifier, qualifierOffset, familyQualifier.length - qualifierOffset, timestamp, + KeyValue.Type.Put, value, 0, value == null ? 0 : value.length); + } + + private static int findFamilyDelimiter(byte[] familyQualifier) { + for (int i = 0; i < familyQualifier.length; i++) { + if (familyQualifier[i] == '\0') { + return i; + } + } + // Keep the failure contract of extractFamilyFromQualifier for malformed responses. + throw new RuntimeException("Cannot get family name"); + } + public static byte[][] extractFamilyFromQualifier(byte[] qualifier) throws Exception { int familyLen = -1; for (int i = 0; i < qualifier.length; i++) { @@ -84,31 +130,8 @@ public static boolean isAllPut(OHOperationType opType, List actio } } - public static void sortHBaseResult(List cells) { - cells.sort(new Comparator() { - @Override - public int compare(Cell cell1, Cell cell2) { - // 1. sort family in lexicographical order - int familyComparison = Bytes.compareTo(cell1.getFamilyArray(), - cell1.getFamilyOffset(), cell1.getFamilyLength(), cell2.getFamilyArray(), - cell2.getFamilyOffset(), cell2.getFamilyLength()); - if (familyComparison != 0) { - return familyComparison; - } - - // 2: sort qualifier in lexicographical order - int qualifierComparison = Bytes.compareTo(cell1.getQualifierArray(), - cell1.getQualifierOffset(), cell1.getQualifierLength(), - cell2.getQualifierArray(), cell2.getQualifierOffset(), - cell2.getQualifierLength()); - if (qualifierComparison != 0) { - return qualifierComparison; - } - - // 3: sort timestamp in descend order - return Long.compare(cell2.getTimestamp(), cell1.getTimestamp()); - } - }); + public static void sortHBaseResult(List cells) { + cells.sort(HBASE_CELL_COMPARATOR); } public static boolean serverCanRetry(ObTableClient tableClient) { diff --git a/src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java b/src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java index f5b11610..2a512c10 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java +++ b/src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java @@ -181,8 +181,7 @@ private void validateOperation(Mutation mt) throws IllegalArgumentException { } if (mt instanceof Put) { // family empty check is in validatePut - HTable.validatePut((Put) mt, maxKeyValueSize); - OHTable.checkFamilyViolation(mt.getFamilyCellMap().keySet(), true); + ohTable.validatePutMutation((Put) mt); } else { OHTable.checkFamilyViolation(mt.getFamilyCellMap().keySet(), false); } diff --git a/src/test/java/com/alipay/oceanbase/hbase/OHTableBatchGetResultTest.java b/src/test/java/com/alipay/oceanbase/hbase/OHTableBatchGetResultTest.java new file mode 100644 index 00000000..39901a93 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/OHTableBatchGetResultTest.java @@ -0,0 +1,142 @@ +/*- + * #%L + * com.oceanbase:obkv-hbase-client + * %% + * Copyright (C) 2022 - 2026 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase; + +import com.alipay.oceanbase.hbase.util.BatchError; +import com.alipay.oceanbase.hbase.result.OHBaseResultCell; +import com.alipay.oceanbase.rpc.ObTableClient; +import com.alipay.oceanbase.rpc.mutation.result.MutationResult; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableSingleOpEntity; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableSingleOpResult; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellBatch; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Row; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; + +public class OHTableBatchGetResultTest { + private ExecutorService executor; + private OHTable table; + + @Before + public void setUp() { + executor = Executors.newSingleThreadExecutor(); + table = new OHTable(Bytes.toBytes("test"), mock(ObTableClient.class), executor); + } + + @After + public void tearDown() { + executor.shutdownNow(); + } + + @Test + public void convertsKqtvAndPreservesRequestOrder() throws Exception { + List actions = Arrays. asList(new Get(Bytes.toBytes("r1")), + new Get(Bytes.toBytes("r2"))); + List raw = Arrays. asList(wrappedResult("r1", "cf\0q1", 2), + wrappedResult("r2", "cf\0q2", 1)); + Object[] results = new Object[2]; + + table.consumePureGetBatchResults(actions, results, raw, new BatchError()); + + assertEquals(2, ((Result) results[0]).size()); + assertEquals(1, ((Result) results[1]).size()); + assertEquals("r1", Bytes.toString(((Result) results[0]).getRow())); + assertEquals("r2", Bytes.toString(((Result) results[1]).getRow())); + assertTrue(((Result) results[0]).rawCells()[0] instanceof OHBaseResultCell); + } + + @Test + public void mapsSingleMissingResultToEmptyResult() throws Exception { + Object[] results = new Object[1]; + table.consumePureGetBatchResults( + Collections. singletonList(new Get(Bytes.toBytes("missing"))), results, + Collections.emptyList(), new BatchError()); + assertTrue(((Result) results[0]).isEmpty()); + } + + @Test(expected = IOException.class) + public void rejectsMalformedKqtvResult() throws Exception { + ObTableSingleOpResult result = new ObTableSingleOpResult(); + result.setEntity(ObTableSingleOpEntity.getInstance(null, null, + new String[] { "K", "Q", "T" }, + new Object[] { Bytes.toBytes("r"), Bytes.toBytes("cf\0q"), 1L })); + table.generateGetResult(result); + } + + @Test + public void consumesCompactKqtvBatch() throws Exception { + ObHBaseCellBatch batch = new ObHBaseCellBatch(2); + batch.setCell(0, Bytes.toBytes("r1"), Bytes.toBytes("cf\0q1"), 100L, Bytes.toBytes("v0")); + batch.setCell(1, Bytes.toBytes("r1"), Bytes.toBytes("cf\0q1"), 99L, Bytes.toBytes("v1")); + ObTableSingleOpEntity entity = new ObTableSingleOpEntity(); + setCompactBatch(entity, batch); + ObTableSingleOpResult result = new ObTableSingleOpResult(); + result.setEntity(entity); + + List cells = table.generateGetResult(result); + + assertEquals(2, cells.size()); + assertEquals("r1", Bytes.toString(cells.get(0).getRowArray(), cells.get(0).getRowOffset(), + cells.get(0).getRowLength())); + assertEquals(99L, cells.get(1).getTimestamp()); + assertTrue(cells.get(0) instanceof OHBaseResultCell); + } + + private static MutationResult wrappedResult(String row, String qualifier, int versions) { + String[] names = new String[versions * 4]; + Object[] values = new Object[versions * 4]; + for (int i = 0; i < versions; i++) { + int offset = i * 4; + names[offset] = "K"; + names[offset + 1] = "Q"; + names[offset + 2] = "T"; + names[offset + 3] = "V"; + values[offset] = Bytes.toBytes(row); + values[offset + 1] = Bytes.toBytes(qualifier); + values[offset + 2] = 100L - i; + values[offset + 3] = Bytes.toBytes("v" + i); + } + ObTableSingleOpResult result = new ObTableSingleOpResult(); + result.setEntity(ObTableSingleOpEntity.getInstance(null, null, names, values)); + return new MutationResult(result); + } + + private static void setCompactBatch(ObTableSingleOpEntity entity, ObHBaseCellBatch batch) + throws Exception { + java.lang.reflect.Field field = ObTableSingleOpEntity.class + .getDeclaredField("hbaseCellBatch"); + field.setAccessible(true); + field.set(entity, batch); + } +} diff --git a/src/test/java/com/alipay/oceanbase/hbase/OHTableCompactPutCellTest.java b/src/test/java/com/alipay/oceanbase/hbase/OHTableCompactPutCellTest.java new file mode 100644 index 00000000..e44f8841 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/OHTableCompactPutCellTest.java @@ -0,0 +1,89 @@ +/*- + * #%L + * OBKV HBase Client Framework + * %% + * Copyright (C) 2022 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase; + +import com.alipay.oceanbase.rpc.ObTableClient; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.OHOperationType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObHbaseRequest; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; + +public class OHTableCompactPutCellTest { + private ExecutorService executorService; + + @Before + public void setUp() { + executorService = Executors.newSingleThreadExecutor(); + } + + @After + public void tearDown() { + executorService.shutdownNow(); + } + + @Test + public void testPutAlwaysUsesCompactCellsWithoutTtl() throws Exception { + Put put = new Put(Bytes.toBytes("row")); + put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q1"), 1001L, Bytes.toBytes("value1")); + put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("qualifier-2"), 1002L, + Bytes.toBytes("value-2")); + + ObHbaseRequest request = newTable().buildHbaseRequest(Collections.singletonList(put), + OHOperationType.PUT); + + assertEquals(1, request.getCfRows().size()); + assertTrue(request.getCfRows().get(0).hasCompactCells()); + assertTrue(request.encode().length > 0); + } + + @Test + public void testPutAlwaysUsesCompactCellsWithTtlAndMultipleRows() throws Exception { + Put first = new Put(Bytes.toBytes("row-1")); + first.setTTL(60000L); + first.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q1"), 2001L, Bytes.toBytes("value-1")); + first.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q2"), 2002L, Bytes.toBytes("value-2")); + first.addColumn(Bytes.toBytes("cf2"), Bytes.toBytes("q3"), 2003L, Bytes.toBytes("value-3")); + Put second = new Put(Bytes.toBytes("row-2")); + second.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q4"), 3001L, Bytes.toBytes("value-4")); + + ObHbaseRequest compact = newTable().buildHbaseRequest(Arrays.asList(first, second), + OHOperationType.PUT_LIST); + + assertEquals(2, compact.getCfRows().size()); + for (int i = 0; i < compact.getCfRows().size(); i++) { + assertTrue(compact.getCfRows().get(i).hasCompactCells()); + } + assertTrue(compact.encode().length > 0); + } + + private OHTable newTable() { + return new OHTable(Bytes.toBytes("test"), mock(ObTableClient.class), executorService); + } +} diff --git a/src/test/java/com/alipay/oceanbase/hbase/OHTableGetMaxRowResultTest.java b/src/test/java/com/alipay/oceanbase/hbase/OHTableGetMaxRowResultTest.java new file mode 100644 index 00000000..ae5e63b4 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/OHTableGetMaxRowResultTest.java @@ -0,0 +1,400 @@ +/*- + * #%L + * com.oceanbase:obkv-hbase-client + * %% + * Copyright (C) 2022 - 2026 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase; + +import com.alipay.oceanbase.rpc.ObTableClient; +import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationLevel; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjMeta; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableSingleOpEntity; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableSingleOpResult; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.AbstractQueryStreamResult; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellBatch; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQueryResult; +import com.alipay.oceanbase.hbase.result.OHBaseResultCell; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.*; +import static org.mockito.Mockito.*; + +public class OHTableGetMaxRowResultTest { + + private OHTable table; + private ExecutorService executorService; + + @Before + public void setUp() { + executorService = Executors.newSingleThreadExecutor(); + table = new OHTable(Bytes.toBytes("test"), mock(ObTableClient.class), executorService); + } + + @After + public void tearDown() { + executorService.shutdownNow(); + } + + @Test + public void testPointGetValidatesOnlyFirstCellAndUsesExpectedRowKey() throws Exception { + byte[] expectedRowKey = Bytes.toBytes("row-1"); + AbstractQueryStreamResult streamResult = stream(row("row-1", "q1", 3L, "v1"), + row("unexpected-later-row", "q2", 2L, "v2")); + List keyValues = new ArrayList<>(); + + boolean found = invokeFillPointGet(streamResult, keyValues, false, Bytes.toBytes("f"), + expectedRowKey, false); + + assertTrue(found); + assertEquals(2, keyValues.size()); + assertArrayEquals(expectedRowKey, CellUtil.cloneRow(keyValues.get(0))); + assertArrayEquals(expectedRowKey, CellUtil.cloneRow(keyValues.get(1))); + assertTrue(keyValues.get(0) instanceof OHBaseResultCell); + verify(streamResult, times(2)).getRow(); + } + + @Test + public void testPointGetRejectsUnexpectedFirstRowKey() throws Exception { + AbstractQueryStreamResult streamResult = stream(row("actual", "q1", 1L, "v1")); + + try { + invokeFillPointGet(streamResult, new ArrayList(), false, Bytes.toBytes("f"), + Bytes.toBytes("expected"), false); + fail("unexpected first rowkey must fail the point Get"); + } catch (InvocationTargetException e) { + assertTrue(e.getCause() instanceof ObTableUnexpectedException); + } + } + + @Test + public void testPointGetConsumesCompactBatchWithoutMaterializingRows() throws Exception { + byte[] expectedRowKey = Bytes.toBytes("row-1"); + AbstractQueryStreamResult streamResult = compactStream(compactBatch( + row("row-1", "q1", 3L, "v1"), row("unexpected-later-row", "q2", 2L, "v2"))); + List keyValues = new ArrayList(); + + boolean found = invokeFillPointGet(streamResult, keyValues, false, Bytes.toBytes("f"), + expectedRowKey, false); + + assertTrue(found); + assertEquals(2, keyValues.size()); + assertArrayEquals(expectedRowKey, CellUtil.cloneRow(keyValues.get(0))); + assertArrayEquals(expectedRowKey, CellUtil.cloneRow(keyValues.get(1))); + assertArrayEquals(Bytes.toBytes("q2"), CellUtil.cloneQualifier(keyValues.get(1))); + verify(streamResult, never()).getRow(); + } + + @Test + public void testPointGetExistenceOnlyDoesNotCreateKeyValue() throws Exception { + AbstractQueryStreamResult streamResult = mock(AbstractQueryStreamResult.class); + when(streamResult.next()).thenReturn(true); + List keyValues = new ArrayList<>(); + + boolean found = invokeFillPointGet(streamResult, keyValues, false, Bytes.toBytes("f"), + Bytes.toBytes("row-1"), true); + + assertTrue(found); + assertTrue(keyValues.isEmpty()); + verify(streamResult, never()).getRow(); + } + + @Test + public void testPointGetReturnsFalseForEmptyResult() throws Exception { + AbstractQueryStreamResult streamResult = mock(AbstractQueryStreamResult.class); + when(streamResult.next()).thenReturn(false); + + boolean found = invokeFillPointGet(streamResult, new ArrayList(), false, + Bytes.toBytes("f"), Bytes.toBytes("row-1"), false); + + assertFalse(found); + } + + @Test + public void testClosestRowBeforeKeepsOnlyCurrentMaxRow() throws Exception { + AbstractQueryStreamResult streamResult = stream(row("row-1", "q1", 4L, "v1"), + row("row-3", "q1", 3L, "v2"), row("row-3", "q2", 2L, "v3"), + row("row-2", "q1", 1L, "v4")); + List keyValues = new ArrayList<>(); + + boolean found = invokeGetMaxRow(streamResult, keyValues, false, Bytes.toBytes("f"), false); + + assertTrue(found); + assertEquals(2, keyValues.size()); + assertArrayEquals(Bytes.toBytes("row-3"), CellUtil.cloneRow(keyValues.get(0))); + assertArrayEquals(Bytes.toBytes("row-3"), CellUtil.cloneRow(keyValues.get(1))); + assertArrayEquals(Bytes.toBytes("q1"), CellUtil.cloneQualifier(keyValues.get(0))); + assertArrayEquals(Bytes.toBytes("q2"), CellUtil.cloneQualifier(keyValues.get(1))); + } + + @Test + public void testClosestRowBeforeExistenceOnlyDoesNotCreateKeyValue() throws Exception { + AbstractQueryStreamResult streamResult = mock(AbstractQueryStreamResult.class); + when(streamResult.next()).thenReturn(true); + List keyValues = new ArrayList<>(); + + boolean found = invokeGetMaxRow(streamResult, keyValues, false, Bytes.toBytes("f"), true); + + assertTrue(found); + assertTrue(keyValues.isEmpty()); + verify(streamResult, never()).getRow(); + } + + @Test + public void testClosestRowBeforeConsumesCompactBatch() throws Exception { + AbstractQueryStreamResult streamResult = compactStream(compactBatch( + row("row-1", "q1", 4L, "v1"), row("row-3", "q1", 3L, "v2"), + row("row-3", "q2", 2L, "v3"), row("row-2", "q1", 1L, "v4"))); + List keyValues = new ArrayList(); + + boolean found = invokeGetMaxRow(streamResult, keyValues, false, Bytes.toBytes("f"), false); + + assertTrue(found); + assertEquals(2, keyValues.size()); + assertArrayEquals(Bytes.toBytes("row-3"), CellUtil.cloneRow(keyValues.get(0))); + assertArrayEquals(Bytes.toBytes("q2"), CellUtil.cloneQualifier(keyValues.get(1))); + verify(streamResult, never()).getRow(); + } + + @Test + public void testTableGroupSplitsFamilyAndQualifier() throws Exception { + byte[] familyAndQualifier = Bytes.add(Bytes.toBytes("family"), new byte[] { 0 }, + Bytes.toBytes("qualifier")); + AbstractQueryStreamResult streamResult = stream(row(Bytes.toBytes("row-1"), + familyAndQualifier, 1L, Bytes.toBytes("value"))); + List keyValues = new ArrayList<>(); + + boolean found = invokeFillPointGet(streamResult, keyValues, true, new byte[0], + Bytes.toBytes("row-1"), false); + + assertTrue(found); + assertEquals(1, keyValues.size()); + assertArrayEquals(Bytes.toBytes("family"), CellUtil.cloneFamily(keyValues.get(0))); + assertArrayEquals(Bytes.toBytes("qualifier"), CellUtil.cloneQualifier(keyValues.get(0))); + assertSame(keyValues.get(0).getFamilyArray(), keyValues.get(0).getQualifierArray()); + } + + @Test + public void testTableGroupSupportsEmptyQualifier() throws Exception { + byte[] familyAndQualifier = Bytes.add(Bytes.toBytes("family"), new byte[] { 0 }); + AbstractQueryStreamResult streamResult = stream(row(Bytes.toBytes("row-1"), + familyAndQualifier, 1L, Bytes.toBytes("value"))); + List keyValues = new ArrayList<>(); + + boolean found = invokeFillPointGet(streamResult, keyValues, true, new byte[0], + Bytes.toBytes("row-1"), false); + + assertTrue(found); + assertEquals(1, keyValues.size()); + assertArrayEquals(Bytes.toBytes("family"), CellUtil.cloneFamily(keyValues.get(0))); + assertArrayEquals(new byte[0], CellUtil.cloneQualifier(keyValues.get(0))); + } + + @Test + public void testTableGroupSupportsMaximumFamilyLength() throws Exception { + byte[] family = new byte[Byte.MAX_VALUE]; + Arrays.fill(family, (byte) 'f'); + byte[] qualifier = new byte[] { 0, (byte) 0xff, 1 }; + byte[] familyAndQualifier = Bytes.add(family, new byte[] { 0 }, qualifier); + AbstractQueryStreamResult streamResult = stream(row(new byte[] { 0, (byte) 0xff }, + familyAndQualifier, 1L, Bytes.toBytes("value"))); + List keyValues = new ArrayList<>(); + + boolean found = invokeGetMaxRow(streamResult, keyValues, true, new byte[0], false); + + assertTrue(found); + assertEquals(1, keyValues.size()); + assertArrayEquals(family, CellUtil.cloneFamily(keyValues.get(0))); + assertArrayEquals(qualifier, CellUtil.cloneQualifier(keyValues.get(0))); + } + + @Test + public void testTableGroupRejectsMissingFamilyDelimiter() throws Exception { + AbstractQueryStreamResult streamResult = stream(row(Bytes.toBytes("row-1"), + Bytes.toBytes("family-without-delimiter"), 1L, Bytes.toBytes("value"))); + + try { + invokeFillPointGet(streamResult, new ArrayList(), true, new byte[0], + Bytes.toBytes("row-1"), false); + fail("missing family delimiter must fail the TableGroup Get"); + } catch (InvocationTargetException e) { + assertTrue(e.getCause() instanceof RuntimeException); + assertEquals("Cannot get family name", e.getCause().getMessage()); + } + } + + @Test + public void testBatchGetBuildsTableGroupKeyValuesFromCompositeQualifier() throws Exception { + byte[] firstFamilyQualifier = Bytes.add(Bytes.toBytes("f1"), new byte[] { 0 }, + Bytes.toBytes("q1")); + byte[] secondFamilyQualifier = Bytes.add(Bytes.toBytes("f2"), new byte[] { 0 }, + Bytes.toBytes("q2")); + ObTableSingleOpEntity entity = mock(ObTableSingleOpEntity.class); + when(entity.getPropertiesValues()).thenReturn( + Arrays.asList(ObObj.getInstance(Bytes.toBytes("row-1")), + ObObj.getInstance(firstFamilyQualifier), ObObj.getInstance(2L), + ObObj.getInstance(Bytes.toBytes("v1")), ObObj.getInstance(Bytes.toBytes("row-1")), + ObObj.getInstance(secondFamilyQualifier), ObObj.getInstance(1L), + ObObj.getInstance(Bytes.toBytes("v2")))); + ObTableSingleOpResult result = mock(ObTableSingleOpResult.class); + when(result.getEntity()).thenReturn(entity); + + List cells = invokeGenerateGetResult(result); + + assertEquals(2, cells.size()); + assertTrue(cells.get(0) instanceof OHBaseResultCell); + assertArrayEquals(Bytes.toBytes("f1"), CellUtil.cloneFamily(cells.get(0))); + assertArrayEquals(Bytes.toBytes("q1"), CellUtil.cloneQualifier(cells.get(0))); + assertArrayEquals(Bytes.toBytes("f2"), CellUtil.cloneFamily(cells.get(1))); + assertArrayEquals(Bytes.toBytes("q2"), CellUtil.cloneQualifier(cells.get(1))); + } + + @Test + public void testQueryAndMutateResultConsumesCompactBatch() throws Exception { + ObTableQueryResult queryResult = compactQueryResult(row("row-1", "q1", 2L, "v1"), + row("row-1", "q2", 1L, "v2")); + List keyValues = new ArrayList(); + + Method method = OHTable.class.getDeclaredMethod("addQueryResultToKeyValueList", + ObTableQueryResult.class, List.class, byte[].class); + method.setAccessible(true); + method.invoke(table, queryResult, keyValues, Bytes.toBytes("f")); + + assertEquals(2, keyValues.size()); + assertArrayEquals(Bytes.toBytes("q1"), CellUtil.cloneQualifier(keyValues.get(0))); + assertArrayEquals(Bytes.toBytes("v2"), CellUtil.cloneValue(keyValues.get(1))); + assertTrue(queryResult.hasHBaseCellBatch()); + } + + @SuppressWarnings("unchecked") + private List invokeGenerateGetResult(ObTableSingleOpResult result) throws Exception { + Method method = OHTable.class.getDeclaredMethod("generateGetResult", + ObTableSingleOpResult.class); + method.setAccessible(true); + return (List) method.invoke(table, result); + } + + private boolean invokeFillPointGet(AbstractQueryStreamResult streamResult, + List keyValues, boolean isTableGroup, byte[] family, + byte[] expectedRowKey, boolean checkExistenceOnly) + throws Exception { + return invokeFillPointGet(table, streamResult, keyValues, isTableGroup, family, + expectedRowKey, checkExistenceOnly); + } + + private boolean invokeFillPointGet(OHTable targetTable, AbstractQueryStreamResult streamResult, + List keyValues, boolean isTableGroup, byte[] family, + byte[] expectedRowKey, boolean checkExistenceOnly) + throws Exception { + Method method = OHTable.class.getDeclaredMethod("fillPointGetFromResult", + AbstractQueryStreamResult.class, List.class, boolean.class, byte[].class, byte[].class, + boolean.class); + method.setAccessible(true); + return (Boolean) method.invoke(targetTable, streamResult, keyValues, isTableGroup, family, + expectedRowKey, checkExistenceOnly); + } + + private boolean invokeGetMaxRow(AbstractQueryStreamResult streamResult, List keyValues, + boolean isTableGroup, byte[] family, boolean checkExistenceOnly) + throws Exception { + Method method = OHTable.class + .getDeclaredMethod("getMaxRowFromResult", AbstractQueryStreamResult.class, List.class, + boolean.class, byte[].class, boolean.class); + method.setAccessible(true); + return (Boolean) method.invoke(table, streamResult, keyValues, isTableGroup, family, + checkExistenceOnly); + } + + private static AbstractQueryStreamResult stream(List... rows) throws Exception { + AbstractQueryStreamResult streamResult = mock(AbstractQueryStreamResult.class); + Boolean[] remaining = new Boolean[Math.max(0, rows.length - 1)]; + Arrays.fill(remaining, true); + when(streamResult.next()).thenReturn(true, remaining).thenReturn(false); + when(streamResult.getRow()).thenReturn(rows[0], Arrays.copyOfRange(rows, 1, rows.length)); + return streamResult; + } + + private static AbstractQueryStreamResult compactStream(ObHBaseCellBatch batch) + throws Exception { + AbstractQueryStreamResult streamResult = mock(AbstractQueryStreamResult.class); + AtomicInteger index = new AtomicInteger(-1); + when(streamResult.next()).thenAnswer(invocation -> index.incrementAndGet() < batch.size()); + when(streamResult.isCurrentHBaseCell()).thenReturn(true); + when(streamResult.getCurrentHBaseCellBatch()).thenReturn(batch); + when(streamResult.getCurrentHBaseCellIndex()).thenAnswer(invocation -> index.get()); + return streamResult; + } + + @SafeVarargs + private static ObHBaseCellBatch compactBatch(List... rows) { + return compactQueryResult(rows).getHBaseCellBatch(); + } + + @SafeVarargs + private static ObTableQueryResult compactQueryResult(List... rows) { + ObTableQueryResult encodedResult = new ObTableQueryResult(); + encodedResult.addPropertiesName("K"); + encodedResult.addPropertiesName("Q"); + encodedResult.addPropertiesName("T"); + encodedResult.addPropertiesName("V"); + encodedResult.addAllPropertiesRows(Arrays.asList(rows)); + encodedResult.setRowCount(rows.length); + + ByteBuf buf = Unpooled.wrappedBuffer(encodedResult.encode()); + try { + ObTableQueryResult decodedResult = new ObTableQueryResult(); + decodedResult.decode(buf); + assertTrue(decodedResult.hasHBaseCellBatch()); + return decodedResult; + } finally { + buf.release(); + } + } + + private static List row(String rowKey, String qualifier, long timestamp, String value) { + return row(Bytes.toBytes(rowKey), Bytes.toBytes(qualifier), timestamp, Bytes.toBytes(value)); + } + + private static List row(byte[] rowKey, byte[] qualifier, long timestamp, byte[] value) { + return Arrays.asList(binaryObj(rowKey), binaryObj(qualifier), new ObObj(new ObObjMeta( + ObObjType.ObInt64Type, ObCollationLevel.CS_LEVEL_NUMERIC, + ObCollationType.CS_TYPE_BINARY, (byte) 0), timestamp), binaryObj(value)); + } + + private static ObObj binaryObj(byte[] value) { + return new ObObj(new ObObjMeta(ObObjType.ObVarcharType, ObCollationLevel.CS_LEVEL_EXPLICIT, + ObCollationType.CS_TYPE_BINARY, (byte) 0), value); + } +} diff --git a/src/test/java/com/alipay/oceanbase/hbase/OHTablePutSkipCellCloneTest.java b/src/test/java/com/alipay/oceanbase/hbase/OHTablePutSkipCellCloneTest.java new file mode 100644 index 00000000..1f0735af --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/OHTablePutSkipCellCloneTest.java @@ -0,0 +1,161 @@ +/*- + * #%L + * OBKV HBase Client Framework + * %% + * Copyright (C) 2022 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase; + +import com.alipay.oceanbase.rpc.ObTableClient; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.OHOperationType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObHbaseCfRows; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObHbaseRequest; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperation; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperationType; +import com.alipay.oceanbase.rpc.util.ObBytesString; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class OHTablePutSkipCellCloneTest { + + private ExecutorService executorService; + + @Before + public void setUp() { + executorService = Executors.newSingleThreadExecutor(); + } + + @After + public void tearDown() { + executorService.shutdownNow(); + } + + @Test + public void testShareContiguousReturnsByteArray() { + byte[] q = Bytes.toBytes("qual"); + byte[] v = Bytes.toBytes("val"); + Cell bare = mock(Cell.class); + when(bare.getQualifierArray()).thenReturn(q); + when(bare.getQualifierOffset()).thenReturn(0); + when(bare.getQualifierLength()).thenReturn(q.length); + when(bare.getValueArray()).thenReturn(v); + when(bare.getValueOffset()).thenReturn(0); + when(bare.getValueLength()).thenReturn(v.length); + Object qObj = OHTable.bytesForPutCell(bare, true, true); + Object vObj = OHTable.bytesForPutCell(bare, true, false); + assertTrue(qObj instanceof byte[]); + assertTrue(vObj instanceof byte[]); + assertSame(q, qObj); + assertSame(v, vObj); + } + + @Test + public void testShareSliceReturnsObBytesStringView() { + byte[] qBuf = new byte[16]; + byte[] q = Bytes.toBytes("qual"); + System.arraycopy(q, 0, qBuf, 3, q.length); + byte[] vBuf = new byte[16]; + byte[] v = Bytes.toBytes("val"); + System.arraycopy(v, 0, vBuf, 2, v.length); + Cell sliced = new KeyValue(Bytes.toBytes("row"), 0, 3, Bytes.toBytes("cf"), 0, 2, qBuf, 3, + q.length, System.currentTimeMillis(), KeyValue.Type.Put, vBuf, 2, v.length); + ObBytesString qView = (ObBytesString) OHTable.bytesForPutCell(sliced, true, true); + ObBytesString vView = (ObBytesString) OHTable.bytesForPutCell(sliced, true, false); + assertSame(sliced.getQualifierArray(), qView.bytes); + assertSame(sliced.getValueArray(), vView.bytes); + assertEquals(sliced.getQualifierOffset(), qView.offset); + assertEquals(sliced.getValueOffset(), vView.offset); + assertEquals(q.length, qView.length()); + assertEquals(v.length, vView.length()); + } + + @Test + public void testShareDisabledAlwaysClones() { + byte[] q = Bytes.toBytes("qual"); + byte[] v = Bytes.toBytes("val"); + Cell contiguous = contiguousCell(Bytes.toBytes("row"), Bytes.toBytes("cf"), q, v); + Object qObj = OHTable.bytesForPutCell(contiguous, false, true); + Object vObj = OHTable.bytesForPutCell(contiguous, false, false); + assertTrue(qObj instanceof byte[]); + assertTrue(vObj instanceof byte[]); + assertNotSame(q, qObj); + assertNotSame(v, vObj); + assertTrue(Bytes.equals(q, (byte[]) qObj)); + assertTrue(Bytes.equals(v, (byte[]) vObj)); + } + + @Test + public void testBuildHbaseRequestSharesViewForSynchronousPut() throws Exception { + OHTable table = new OHTable(Bytes.toBytes("t"), mock(ObTableClient.class), executorService); + byte[] q = Bytes.toBytes("q1"); + byte[] v = Bytes.toBytes("v1"); + Put put = new Put(Bytes.toBytes("row")); + put.addColumn(Bytes.toBytes("cf"), q, v); + Cell src = put.getFamilyCellMap().get(Bytes.toBytes("cf")).get(0); + + ObHbaseRequest request = table.buildHbaseRequest(Collections.singletonList(put), + OHOperationType.PUT); + ObHbaseCfRows cfRows = request.getCfRows().get(0); + assertTrue(cfRows.hasCompactCells()); + assertSame(src.getQualifierArray(), compactByteArrays(cfRows, "compactQualifierArrays")[0]); + assertSame(src.getValueArray(), compactByteArrays(cfRows, "compactValueArrays")[0]); + } + + @Test + public void testLegacyTtlReusesSingleValueClone() throws Exception { + OHTable table = new OHTable(Bytes.toBytes("t"), mock(ObTableClient.class), executorService); + byte[] value = Bytes.toBytes("payload"); + KeyValue kv = new KeyValue(Bytes.toBytes("row"), Bytes.toBytes("cf"), Bytes.toBytes("q"), + value); + long ttl = 60_000L; + ObTableOperation op = table.buildObTableOperation(kv, ObTableOperationType.INSERT_OR_UPDATE, + ttl); + assertTrue(op != null); + java.lang.reflect.Method buildMutation = OHTable.class.getDeclaredMethod("buildMutation", + Cell.class, ObTableOperationType.class, boolean.class, byte[].class, Long.class); + buildMutation.setAccessible(true); + Object mutation = buildMutation.invoke(table, kv, ObTableOperationType.INSERT_OR_UPDATE, + false, null, ttl); + assertTrue(mutation != null); + } + + private static Cell contiguousCell(byte[] row, byte[] family, byte[] qualifier, byte[] value) { + return new KeyValue(row, 0, row.length, family, 0, family.length, qualifier, 0, + qualifier.length, System.currentTimeMillis(), KeyValue.Type.Put, value, 0, value.length); + } + + private static byte[][] compactByteArrays(ObHbaseCfRows cfRows, String fieldName) + throws Exception { + Field field = ObHbaseCfRows.class.getDeclaredField(fieldName); + field.setAccessible(true); + return (byte[][]) field.get(cfRows); + } +} diff --git a/src/test/java/com/alipay/oceanbase/hbase/OHTablePutValidationTest.java b/src/test/java/com/alipay/oceanbase/hbase/OHTablePutValidationTest.java new file mode 100644 index 00000000..a6d07b9e --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/OHTablePutValidationTest.java @@ -0,0 +1,128 @@ +/*- + * #%L + * OBKV HBase Client Framework + * %% + * Copyright (C) 2022 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase; + +import com.alipay.oceanbase.rpc.ObTableClient; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.KeyValueUtil; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class OHTablePutValidationTest { + + private static final byte[] ROW = Bytes.toBytes("row"); + private static final byte[] FAMILY = Bytes.toBytes("cf"); + + private ExecutorService executorService; + + @Before + public void setUp() { + executorService = Executors.newSingleThreadExecutor(); + } + + @After + public void tearDown() { + executorService.shutdownNow(); + } + + @Test + public void testStaticValidationDoesNotMaterializeDeprecatedFamilyMap() { + Put put = newPut(FAMILY, "q", "value"); + Cell cell = put.getFamilyCellMap().get(FAMILY).get(0); + + OHTable.validatePut(put, KeyValueUtil.length(cell)); + } + + @Test + public void testTableValidationDoesNotMaterializeDeprecatedFamilyMap() { + OHTable table = new OHTable(Bytes.toBytes("test"), mock(ObTableClient.class), + executorService); + + table.validatePutMutation(newPut(FAMILY, "q", "value")); + } + + @Test + public void testCellAtMaximumSizeIsAccepted() { + Put put = newPut(FAMILY, "q", "value"); + Cell cell = put.getFamilyCellMap().get(FAMILY).get(0); + + OHTable.validatePut(put, KeyValueUtil.length(cell)); + } + + @Test + public void testCellOverMaximumSizeIsRejected() { + Put put = newPut(FAMILY, "q", "value"); + Cell cell = put.getFamilyCellMap().get(FAMILY).get(0); + + try { + OHTable.validatePut(put, KeyValueUtil.length(cell) - 1); + fail("oversized cell should fail"); + } catch (IllegalArgumentException expected) { + assertTrue(expected.getMessage().contains("KeyValue size too large")); + } + } + + @Test + public void testEmptyPutIsRejected() { + try { + OHTable.validatePut(new Put(ROW), -1); + fail("empty put should fail"); + } catch (IllegalArgumentException expected) { + assertTrue(expected.getMessage().contains("No columns to insert")); + } + } + + @Test + public void testMultipleFamiliesUseOriginalCellMap() { + Put put = newPut(FAMILY, "q1", "value1"); + put.addColumn(Bytes.toBytes("cf2"), Bytes.toBytes("q2"), Bytes.toBytes("value2")); + + OHTable.validatePut(put, Integer.MAX_VALUE); + } + + @Test + public void testNonKeyValueCellUsesCalculatedLengthWithoutConversion() { + Put put = newPut(FAMILY, "q", "value"); + Cell cell = mock(Cell.class); + when(cell.getRowLength()).thenReturn((short) 3); + when(cell.getFamilyLength()).thenReturn((byte) 2); + when(cell.getQualifierLength()).thenReturn(1); + when(cell.getValueLength()).thenReturn(5); + when(cell.getTagsLength()).thenReturn(0); + put.getFamilyCellMap().get(FAMILY).set(0, cell); + + OHTable.validatePut(put, KeyValueUtil.length(cell)); + } + + private static Put newPut(byte[] family, String qualifier, String value) { + Put put = new Put(ROW); + put.addColumn(family, Bytes.toBytes(qualifier), Bytes.toBytes(value)); + return put; + } +} diff --git a/src/test/java/com/alipay/oceanbase/hbase/result/ClientStreamScannerCompactResultTest.java b/src/test/java/com/alipay/oceanbase/hbase/result/ClientStreamScannerCompactResultTest.java new file mode 100644 index 00000000..79e84030 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/result/ClientStreamScannerCompactResultTest.java @@ -0,0 +1,119 @@ +/*- + * #%L + * OBKV HBase Client Framework + * %% + * Copyright (C) 2022 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase.result; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellBatch; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellRow; +import com.alipay.oceanbase.rpc.stream.ObTableClientQueryAsyncStreamResult; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.Test; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class ClientStreamScannerCompactResultTest { + + @Test + public void testCompactResultUsesKeyValueAndDirectConsume() throws Exception { + ObTableClientQueryAsyncStreamResult streamResult = compactStreamResult(compactRow( + new String[] { "q-2", "q-1" }, new long[] { 101L, 102L })); + Scan scan = new Scan(); + scan.addFamily(bytes("f")); + ClientStreamScanner scanner = new ClientStreamScanner(streamResult, "test", scan, false, + null); + + Result result = scanner.next(); + + assertEquals(2, result.size()); + assertTrue(result.rawCells()[0] instanceof KeyValue); + assertArrayEquals(bytes("q-1"), CellUtil.cloneQualifier(result.rawCells()[0])); + assertArrayEquals(bytes("q-2"), CellUtil.cloneQualifier(result.rawCells()[1])); + verify(streamResult, never()).getRow(); + verify(streamResult, never()).getCacheRows(); + } + + @Test + public void testCompactTableGroupResultUsesQualifierOffsets() throws Exception { + ObTableClientQueryAsyncStreamResult streamResult = compactStreamResult(compactRow( + new String[] { "f1\0q-1", "f2\0q-2" }, new long[] { 102L, 101L })); + ClientStreamScanner scanner = new ClientStreamScanner(streamResult, "test", new Scan(), + true, null); + + Result result = scanner.next(); + + assertEquals(2, result.size()); + assertTrue(result.rawCells()[0] instanceof KeyValue); + assertArrayEquals(bytes("f1"), CellUtil.cloneFamily(result.rawCells()[0])); + assertArrayEquals(bytes("q-1"), CellUtil.cloneQualifier(result.rawCells()[0])); + assertArrayEquals(bytes("f2"), CellUtil.cloneFamily(result.rawCells()[1])); + assertArrayEquals(bytes("q-2"), CellUtil.cloneQualifier(result.rawCells()[1])); + } + + private static ObTableClientQueryAsyncStreamResult compactStreamResult(ObHBaseCellRow row) + throws Exception { + ObTableClientQueryAsyncStreamResult streamResult = mock(ObTableClientQueryAsyncStreamResult.class); + when(streamResult.next()).thenReturn(true); + when(streamResult.isCurrentHBaseCell()).thenReturn(true); + when(streamResult.drainCurrentHBaseRow()).thenReturn(row); + when(streamResult.getTableName()).thenReturn("test"); + return streamResult; + } + + private static ObHBaseCellRow compactRow(String[] qualifiers, long[] timestamps) + throws Exception { + assertEquals(qualifiers.length, timestamps.length); + Constructor batchConstructor = ObHBaseCellBatch.class + .getDeclaredConstructor(int.class); + batchConstructor.setAccessible(true); + ObHBaseCellBatch batch = batchConstructor.newInstance(qualifiers.length); + Method setCell = ObHBaseCellBatch.class.getDeclaredMethod("setCell", int.class, + byte[].class, byte[].class, long.class, byte[].class); + setCell.setAccessible(true); + byte[] rowKey = bytes("row-1"); + for (int i = 0; i < qualifiers.length; i++) { + setCell.invoke(batch, i, rowKey, bytes(qualifiers[i]), timestamps[i], bytes("value-" + + i)); + } + + Constructor rowConstructor = ObHBaseCellRow.class + .getDeclaredConstructor(byte[].class); + rowConstructor.setAccessible(true); + ObHBaseCellRow row = rowConstructor.newInstance(rowKey); + Method addSlice = ObHBaseCellRow.class.getDeclaredMethod("addSlice", + ObHBaseCellBatch.class, int.class, int.class); + addSlice.setAccessible(true); + addSlice.invoke(row, batch, 0, qualifiers.length); + return row; + } + + private static byte[] bytes(String value) { + return Bytes.toBytes(value); + } +} diff --git a/src/test/java/com/alipay/oceanbase/hbase/result/OHBaseResultCellTest.java b/src/test/java/com/alipay/oceanbase/hbase/result/OHBaseResultCellTest.java new file mode 100644 index 00000000..11878b68 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/hbase/result/OHBaseResultCellTest.java @@ -0,0 +1,124 @@ +/*- + * #%L + * OBKV HBase Client Framework + * %% + * Copyright (C) 2022 OceanBase Group + * %% + * OBKV HBase Client Framework is licensed under Mulan PSL v2. + * You can use this software according to the terms and conditions of the Mulan PSL v2. + * You may obtain a copy of Mulan PSL v2 at: + * http://license.coscl.org.cn/MulanPSL2 + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, + * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, + * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. + * See the Mulan PSL v2 for more details. + * #L% + */ + +package com.alipay.oceanbase.hbase.result; + +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.KeyValueUtil; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.Collections; + +import static org.junit.Assert.*; + +public class OHBaseResultCellTest { + + @Test + public void testFieldBackedCellAccessors() { + byte[] row = bytes("row-1"); + byte[] family = bytes("f"); + byte[] qualifier = bytes("q1"); + byte[] value = bytes("value-1"); + OHBaseResultCell cell = OHBaseResultCell.create(row, family, qualifier, 123L, value); + + assertSame(row, cell.getRowArray()); + assertSame(family, cell.getFamilyArray()); + assertSame(qualifier, cell.getQualifierArray()); + assertSame(value, cell.getValueArray()); + assertEquals(0, cell.getRowOffset()); + assertEquals(row.length, cell.getRowLength()); + assertEquals(0, cell.getFamilyOffset()); + assertEquals(family.length, cell.getFamilyLength()); + assertEquals(0, cell.getQualifierOffset()); + assertEquals(qualifier.length, cell.getQualifierLength()); + assertEquals(0, cell.getValueOffset()); + assertEquals(value.length, cell.getValueLength()); + assertEquals(123L, cell.getTimestamp()); + assertEquals(KeyValue.Type.Put.getCode(), cell.getTypeByte()); + assertEquals(0L, cell.getMvccVersion()); + assertEquals(0L, cell.getSequenceId()); + assertEquals(0, cell.getTagsLength()); + + assertArrayEquals(row, cell.getRow()); + assertArrayEquals(family, cell.getFamily()); + assertArrayEquals(qualifier, cell.getQualifier()); + assertArrayEquals(value, cell.getValue()); + assertNotSame(row, cell.getRow()); + assertNotSame(value, cell.getValue()); + } + + @Test + public void testTableGroupCellUsesSharedArrayRanges() { + byte[] familyQualifier = new byte[] { 'f', '1', 0, 'q', '1' }; + OHBaseResultCell cell = OHBaseResultCell.createTableGroup(bytes("row-1"), familyQualifier, + 99L, bytes("v")); + + assertSame(familyQualifier, cell.getFamilyArray()); + assertSame(familyQualifier, cell.getQualifierArray()); + assertEquals(0, cell.getFamilyOffset()); + assertEquals(2, cell.getFamilyLength()); + assertEquals(3, cell.getQualifierOffset()); + assertEquals(2, cell.getQualifierLength()); + assertArrayEquals(bytes("f1"), CellUtil.cloneFamily(cell)); + assertArrayEquals(bytes("q1"), CellUtil.cloneQualifier(cell)); + } + + @Test + public void testResultAndLegacyKeyValueApisRemainCompatible() { + byte[] family = bytes("f"); + byte[] qualifier = bytes("q"); + byte[] value = bytes("value"); + Cell cell = OHBaseResultCell.create(bytes("row"), family, qualifier, 7L, value); + Result result = Result.create(Collections.singletonList(cell)); + + assertSame(cell, result.rawCells()[0]); + assertSame(cell, result.listCells().get(0)); + assertArrayEquals(value, result.getValue(family, qualifier)); + ByteBuffer valueBuffer = result.getValueAsByteBuffer(family, qualifier); + assertArrayEquals(value, Bytes.toBytes(valueBuffer)); + assertSame(cell, result.getColumnLatestCell(family, qualifier)); + assertEquals(1, result.getColumnCells(family, qualifier).size()); + assertArrayEquals(value, result.getFamilyMap(family).get(qualifier)); + + KeyValue converted = KeyValueUtil.ensureKeyValue(cell); + assertArrayEquals(bytes("row"), CellUtil.cloneRow(converted)); + assertArrayEquals(family, CellUtil.cloneFamily(converted)); + assertArrayEquals(qualifier, CellUtil.cloneQualifier(converted)); + assertArrayEquals(value, CellUtil.cloneValue(converted)); + assertEquals(7L, converted.getTimestamp()); + } + + @Test(expected = RuntimeException.class) + public void testTableGroupCellRejectsMissingDelimiter() { + OHBaseResultCell.createTableGroup(bytes("row"), bytes("family-qualifier"), 1L, bytes("v")); + } + + @Test(expected = IllegalArgumentException.class) + public void testCellRejectsOversizedFamily() { + OHBaseResultCell.create(bytes("row"), new byte[Byte.MAX_VALUE + 1], bytes("q"), 1L, + bytes("v")); + } + + private static byte[] bytes(String value) { + return Bytes.toBytes(value); + } +}