diff --git a/src/main/java/com/alipay/oceanbase/hbase/OHTable.java b/src/main/java/com/alipay/oceanbase/hbase/OHTable.java index 676831ba..34221073 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/OHTable.java +++ b/src/main/java/com/alipay/oceanbase/hbase/OHTable.java @@ -23,10 +23,14 @@ import com.alipay.oceanbase.hbase.filter.HBaseFilterUtils; import com.alipay.oceanbase.hbase.result.ClientStreamScanner; import com.alipay.oceanbase.hbase.util.*; +import com.alipay.oceanbase.rpc.ObGlobal; import com.alipay.oceanbase.rpc.ObTableClient; import com.alipay.oceanbase.rpc.exception.ObTableException; +import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException; import com.alipay.oceanbase.rpc.mutation.BatchOperation; import com.alipay.oceanbase.rpc.mutation.result.BatchOperationResult; +import com.alipay.oceanbase.rpc.mutation.result.MutationResult; +import com.alipay.oceanbase.rpc.protocol.payload.ObPayload; import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj; import com.alipay.oceanbase.rpc.protocol.payload.impl.ObRowKey; import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.*; @@ -39,6 +43,7 @@ import com.alipay.oceanbase.rpc.stream.ObTableClientQueryStreamResult; import com.alipay.oceanbase.rpc.table.ObHBaseParams; import com.alipay.oceanbase.rpc.table.ObKVParams; +import com.alipay.oceanbase.rpc.table.ObTableClientQueryImpl; import com.alipay.sofa.common.thread.SofaThreadPoolExecutor; import com.google.protobuf.Descriptors; @@ -68,6 +73,8 @@ import static com.alipay.oceanbase.hbase.util.Preconditions.checkNotNull; import static com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory.LCD; import static com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory.TABLE_HBASE_LOGGER_SPACE; +import static com.alipay.oceanbase.rpc.mutation.MutationFactory.colVal; +import static com.alipay.oceanbase.rpc.mutation.MutationFactory.row; import static com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperation.getInstance; import static com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperationType.*; import static com.alipay.sofa.common.thread.SofaThreadPoolConstants.SOFA_THREAD_POOL_LOGGING_CAPABILITY; @@ -523,8 +530,18 @@ public boolean exists(Get get) throws IOException { @Override public boolean[] existsAll(List gets) throws IOException { boolean[] ret = new boolean[gets.size()]; - for (int i = 0; i < gets.size(); ++i) { - ret[i] = exists(gets.get(i)); + List newGets = new ArrayList<>(); + // 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 + // TODO: adjust ExistOnly in server when using batch get + for (Get get : gets) { + Get newGet = new Get(get); + newGet.setCheckExistenceOnly(false); + newGets.add(newGet); + } + Result[] results = get(newGets); + for (int i = 0; i < results.length; ++i) { + ret[i] = !results[i].isEmpty(); } return ret; } @@ -681,7 +698,7 @@ private void compatOldServerBatch(final List actions, final Objec } @Override - public void batch(final List actions, final Object[] results) throws IOException{ + public void batch(final List actions, final Object[] results) throws IOException { if (actions == null) { return; } @@ -693,7 +710,7 @@ public void batch(final List actions, final Object[] results) thr BatchError batchError = new BatchError(); obTableClient.setRuntimeBatchExecutor(executePool); List resultMapSingleOp = new LinkedList<>(); - if (!CompatibilityUtil.isBatchSupport()) { + if (!ObGlobal.isHBaseBatchSupport()) { try { compatOldServerBatch(actions, results, batchError); } catch (Exception e) { @@ -716,6 +733,23 @@ public void batch(final List actions, final Object[] results) thr 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"); + } + } else { + throw new ObTableUnexpectedException("Unexpected type of result in batch"); + } + } } else { if (results != null) { results[i] = new Result(); @@ -729,17 +763,57 @@ public void batch(final List actions, final Object[] results) thr } } + private List generateGetResult(ObTableSingleOpResult getResult) throws IOException { + List cells = new ArrayList<>(); + 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][]; + try { + // split family and qualifier + familyAndQualifier = OHBaseFuncUtils + .extractFamilyFromQualifier((byte[]) propertiesValues.get(valueIdx + 1).getValue()); + } 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; + } + private String getTargetTableName(List actions) { byte[] family = null; for (Row action : actions) { if (action instanceof RowMutations || action instanceof RegionCoprocessorServiceExec) { throw new FeatureNotSupportedException("not supported yet'"); } else { - Mutation mutation = (Mutation) action; - if (mutation.getFamilyCellMap().size() != 1) { + Set familySet = null; + if (action instanceof Get){ + Get get = (Get) action; + familySet = get.familySet(); + } else { + Mutation mutation = (Mutation) action; + familySet = mutation.getFamilyCellMap().keySet(); + } + if (familySet == null) { + throw new ObTableUnexpectedException("Fail to get family set in action"); + } + if (familySet.size() != 1) { return getTargetTableName(tableNameString); } else { - byte[] nextFamily = mutation.getFamilyCellMap().keySet().iterator().next(); + byte[] nextFamily = familySet.iterator().next(); if (family != null && !Arrays.equals(family, nextFamily)) { return getTargetTableName(tableNameString); } else if (family == null) { @@ -825,8 +899,8 @@ private String getTargetTableName(String tableNameString) { return tableNameString; } - // To enable the server to identify the column family to which a qualifier belongs, - // the client writes the column family name into the qualifier. + // To enable the server to identify the column family to which a qualifier belongs, + // the client writes the column family name into the qualifier. // The server then parses this information to determine the table that needs to be operated on. private void processColumnFilters(NavigableSet columnFilters, Map> familyMap) { @@ -862,8 +936,8 @@ public Result call() throws IOException { if (get.getFamilyMap().keySet() == null || get.getFamilyMap().keySet().isEmpty() || get.getFamilyMap().size() > 1) { - // In a Get operation where the family map is greater than 1 or equal to 0, - // we handle this by appending the column family to the qualifier on the client side. + // In a Get operation where the family map is greater than 1 or equal to 0, + // we handle this by appending the column family to the qualifier on the client side. // The server can then use this information to filter the appropriate column families and qualifiers. if (!get.getColumnFamilyTimeRange().isEmpty()) { throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now"); @@ -920,8 +994,12 @@ public Result call() throws IOException { @Override public Result[] get(List gets) throws IOException { Result[] results = new Result[gets.size()]; - for (int i = 0; i < gets.size(); i++) { - results[i] = get(gets.get(i)); + if (ObGlobal.isHBaseBatchGetSupport()) { // get only supported in BatchSupport version + batch(gets, results); + } else { + for (int i = 0; i < gets.size(); i++) { + results[i] = get(gets.get(i)); + } } return results; } @@ -948,8 +1026,8 @@ public ResultScanner call() throws IOException { if (scan.getFamilyMap().keySet() == null || scan.getFamilyMap().keySet().isEmpty() || scan.getFamilyMap().size() > 1) { - // In a Scan operation where the family map is greater than 1 or equal to 0, - // we handle this by appending the column family to the qualifier on the client side. + // In a Scan operation where the family map is greater than 1 or equal to 0, + // we handle this by appending the column family to the qualifier on the client side. // The server can then use this information to filter the appropriate column families and qualifiers. if (!scan.getColumnFamilyTimeRange().isEmpty()) { throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now"); @@ -1464,7 +1542,6 @@ public void close() throws IOException { if (cleanupPoolOnClose) { executePool.shutdown(); } - ObTableClientManager.clear(); } @Override @@ -1514,6 +1591,8 @@ public void batchCoprocessorService(Descriptors.MethodDescri @Override public void setOperationTimeout(int operationTimeout) { this.operationTimeout = operationTimeout; + this.obTableClient.setRuntimeMaxWait(operationTimeout); + this.obTableClient.setRuntimeBatchMaxWait(operationTimeout); this.operationExecuteInPool = this.configuration.getBoolean( HBASE_CLIENT_OPERATION_EXECUTE_IN_POOL, (this.operationTimeout != HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT)); @@ -1834,21 +1913,62 @@ private KeyValue modifyQualifier(Cell original, byte[] newQualifier) { byte[] value = CellUtil.cloneValue(original); long timestamp = original.getTimestamp(); KeyValue.Type type = KeyValue.Type.codeToType(original.getType().getCode()); - // Create a new KeyValue with the modified qualifier + // Create a new KeyValue with the modified qualifier return new KeyValue(row, family, newQualifier, timestamp, type, value); } private BatchOperation buildBatchOperation(String tableName, List actions, boolean isTableGroup, List resultMapSingleOp) throws FeatureNotSupportedException, - IllegalArgumentException { + IllegalArgumentException, + IOException { BatchOperation batch = obTableClient.batchOperation(tableName); int posInList = -1; int singleOpResultNum; for (Row row : actions) { singleOpResultNum = 0; posInList++; - if (row instanceof Put) { + if (row instanceof Get) { + if (!ObGlobal.isHBaseBatchGetSupport()) { + throw new FeatureNotSupportedException("server does not support batch get"); + } + ++singleOpResultNum; + Get get = (Get) row; + ObTableQuery obTableQuery; + // In a Get operation in ls batch, we need to determine whether the get is a table-group operation or not, + // we handle this by appending the column family to the qualifier on the client side. + // The server can then use this information to filter the appropriate column families and qualifiers. + if ((get.getFamilyMap().keySet().isEmpty() + || get.getFamilyMap().size() > 1) && + !get.getColumnFamilyTimeRange().isEmpty()) { + throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now"); + } else if (get.getFamilyMap().size() == 1 && !get.getColumnFamilyTimeRange().isEmpty()) { + byte[] family = get.getFamilyMap().keySet().iterator().next(); + Map colFamTimeRangeMap = get.getColumnFamilyTimeRange(); + if (colFamTimeRangeMap.size() > 1) { + throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now"); + } else if (colFamTimeRangeMap.get(family) == null) { + throw new IllegalArgumentException("Get family is not matched in ColumnFamilyTimeRange"); + } else { + TimeRange tr = colFamTimeRangeMap.get(family); + get.setTimeRange(tr.getMin(), tr.getMax()); + } + } + NavigableSet columnFilters = new TreeSet<>(Bytes.BYTES_COMPARATOR); + // in batch get, we need to carry family in qualifier to server even this get is a single-cf operation + // because the entire batch may be a multi-cf batch so do not carry family + // family in qualifier helps us to know which table to query + processColumnFilters(columnFilters, get.getFamilyMap()); + obTableQuery = buildObTableQuery(get, columnFilters); + ObTableClientQueryImpl query = new ObTableClientQueryImpl(tableName, obTableQuery, obTableClient); + try { + query.setRowKey(row(colVal("K", Bytes.toString(get.getRow())), colVal("Q", null), colVal("T", null))); + } catch (Exception e) { + logger.error("unexpected error occurs when set row key", e); + throw new IOException(e); + } + batch.addOperation(query); + } else if (row instanceof Put) { Put put = (Put) row; if (put.isEmpty()) { throw new IllegalArgumentException("No columns to insert for #" @@ -1882,7 +2002,7 @@ private BatchOperation buildBatchOperation(String tableName, List } } else { throw new FeatureNotSupportedException( - "not supported other type in batch yet,only support put and delete"); + "not supported other type in batch yet,only support get, put and delete"); } resultMapSingleOp.add(singleOpResultNum); } @@ -1983,8 +2103,8 @@ public static void checkFamilyViolation(Collection families, boolean che } } - // This method is currently only used for append and increment operations. - // It restricts these two methods to use multi-column family operations. + // This method is currently only used for append and increment operations. + // It restricts these two methods to use multi-column family operations. // Note: After completing operations on multiple column families, they are deleted using the method described above. public static void checkFamilyViolationForOneFamily(Collection families) { if (families == null || families.size() == 0) { diff --git a/src/main/java/com/alipay/oceanbase/hbase/util/CompatibilityUtil.java b/src/main/java/com/alipay/oceanbase/hbase/util/CompatibilityUtil.java deleted file mode 100644 index 39db637c..00000000 --- a/src/main/java/com/alipay/oceanbase/hbase/util/CompatibilityUtil.java +++ /dev/null @@ -1,9 +0,0 @@ -package com.alipay.oceanbase.hbase.util; - -import com.alipay.oceanbase.rpc.ObGlobal; - -public class CompatibilityUtil { - public static boolean isBatchSupport() { - return ObGlobal.OB_VERSION > ObGlobal.OB_VERSION_4_3_4_0; - } -} diff --git a/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java b/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java index 34d3d3b3..5db940aa 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java +++ b/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java @@ -126,18 +126,6 @@ public static ObTableClient getOrCreateObTableClient(ObTableClientKey obTableCli return OB_TABLE_CLIENT_INSTANCE.get(obTableClientKey); } - public static void clear() throws IOException { - try { - for (Map.Entry pair : OB_TABLE_CLIENT_INSTANCE.entrySet()) { - pair.getValue().close(); - } - } - catch (Exception e) { - throw new IOException("fail to close tableClient" , e); - } - OB_TABLE_CLIENT_INSTANCE.clear(); - } - public static class ObTableClientKey { private String paramUrl; private String fullUserName; diff --git a/src/test/java/com/alipay/oceanbase/hbase/HTableMultiCFTestBase.java b/src/test/java/com/alipay/oceanbase/hbase/HTableMultiCFTestBase.java index 83236172..a22bc769 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/HTableMultiCFTestBase.java +++ b/src/test/java/com/alipay/oceanbase/hbase/HTableMultiCFTestBase.java @@ -826,6 +826,426 @@ public void update(byte[] region, byte[] row, Result result) { }); assertEquals(11, updateCounter[0]); + /*--------------------------------------test batch get------------------------------------------*/ + // single-cf test + batchLsit.clear(); + for (int i = 0; i < rows; ++i) { + Put put = new Put(toBytes("Key" + i)); + delete = new Delete(toBytes("Key" + i)); + batchLsit.add(delete); + put.addColumn(family1, family1_column1, family1_value); + put.addColumn(family1, family1_column2, family1_value); + put.addColumn(family1, family1_column3, family1_value); + put.addColumn(family2, family2_column1, family2_value); + put.addColumn(family2, family2_column2, family2_value); + put.addColumn(family3, family3_column1, family3_value); + batchLsit.add(put); + } + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + batchLsit.clear(); + // get + Get get1 = new Get("Key1".getBytes()); + get1.addFamily(family1); + batchLsit.add(get1); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(1, results.length); + Result getResult = (Result) results[0]; + Assert.assertEquals(3, getResult.rawCells().length); + for (Cell keyValue : getResult.rawCells()) { + System.out.printf("1. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + Assert.assertTrue(Bytes.toString(CellUtil.cloneFamily(keyValue)).equals("family_with_group1")); + } + // delete + get + delete = new Delete(toBytes("Key1")); + batchLsit.clear(); + batchLsit.add(delete); + batchLsit.add(get1); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(2, results.length); + Assert.assertEquals(0, ((Result) results[1]).rawCells().length); + + // put + get + Put put = new Put(toBytes("Key1")); + put.addColumn(family1, family1_column1, family1_value); + put.addColumn(family1, family1_column2, family2_value); + put.addColumn(family1, family1_column3, family3_value); + batchLsit.clear(); + batchLsit.add(put); + batchLsit.add(get1); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(2, results.length); + getResult = (Result) results[1]; + Assert.assertEquals(3, getResult.rawCells().length); + for (Cell keyValue : getResult.rawCells()) { + System.out.printf("2. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + + // put + delete + get + Get get2 = new Get("Key2".getBytes()); + get1.setMaxVersions(10); + get2.addColumn(family1, family1_column1); + batchLsit.clear(); + batchLsit.add(delete); + batchLsit.add(put); + batchLsit.add(get1); + batchLsit.add(delete); + batchLsit.add(get1); + batchLsit.add(get2); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(6, results.length); + Result key1Result = (Result) results[2]; + Assert.assertEquals(3, key1Result.rawCells().length); + for (Cell keyValue : key1Result.rawCells()) { + System.out.printf("3. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + Result empResult = (Result) results[4]; + Assert.assertEquals(0, empResult.rawCells().length); + + Result key2Result = (Result) results[5]; + Assert.assertEquals(1, key2Result.rawCells().length); + for (Cell keyValue : key2Result.rawCells()) { + System.out.printf("4. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + + // set maxVersion + put = new Put(toBytes("Key2")); + put.addColumn(family1, family1_column1, family1_value); + get2.setMaxVersions(10); + batchLsit.clear(); + batchLsit.add(put); + batchLsit.add(get2); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(2, results.length); + key2Result = (Result) results[1]; + Assert.assertEquals(2, key2Result.rawCells().length); + for (Cell keyValue : key2Result.rawCells()) { + System.out.printf("5. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + get2.setMaxVersions(2); + Get get2All = new Get(toBytes("Key2")); + get2All.addColumn(family1, family1_column1); + get2All.setMaxVersions(); + batchLsit.clear(); + batchLsit.add(put); + batchLsit.add(get2); + batchLsit.add(get2All); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(3, results.length); + key2Result = (Result) results[1]; + Result key2AllResult = (Result) results[2]; + Assert.assertEquals(3, key2AllResult.getColumnCells(family1, family1_column1).size()); + Assert.assertEquals(2, key2Result.getColumnCells(family1, family1_column1).size()); + for (Cell keyValue : key2Result.rawCells()) { + System.out.printf("6. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + + // compare with get interface + Result key2GetResult = multiCfHTable.get(get2); + Assert.assertEquals(key2GetResult.rawCells().length, key2Result.rawCells().length); + for (int i = 0; i < key2GetResult.size(); ++i) { + Assert.assertEquals(Bytes.toString(key2Result.getRow()), + Bytes.toString(key2GetResult.getRow())); + Assert.assertEquals(Bytes.toString(CellUtil.cloneFamily(key2Result.rawCells()[i])), + Bytes.toString(CellUtil.cloneFamily(key2GetResult.rawCells()[i]))); + Assert.assertEquals(Bytes.toString(CellUtil.cloneQualifier(key2Result.rawCells()[i])), + Bytes.toString(CellUtil.cloneQualifier(key2GetResult.rawCells()[i]))); + Assert.assertEquals(key2Result.rawCells()[i].getTimestamp(), + key2GetResult.rawCells()[i].getTimestamp()); + Assert.assertEquals(Bytes.toString(CellUtil.cloneValue(key2Result.rawCells()[i])), + Bytes.toString(CellUtil.cloneValue(key2GetResult.rawCells()[i]))); + } + // test batch get + batchLsit.clear(); + for (int i = 0; i < rows; ++i) { + get = new Get(toBytes("Key" + i)); + get.addFamily(family1); + get.setMaxVersions(10); + batchLsit.add(get); + } + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(rows, results.length); + for (int i = 0; i < rows; ++i) { + get = new Get(toBytes("Key" + i)); + get.addFamily(family1); + get.setMaxVersions(10); + getResult = multiCfHTable.get(get); + Result batchGetResult = (Result) results[i]; + if (Bytes.toString(get.getRow()).equals("Key1")) { + Assert.assertEquals(0, batchGetResult.rawCells().length); + } else if (Bytes.toString(get.getRow()).equals("Key2")) { + // 3 + 1 + 1 + Assert.assertEquals(5, batchGetResult.rawCells().length); + } else { + Assert.assertEquals(3, batchGetResult.rawCells().length); + } + Assert.assertEquals(getResult.rawCells().length, batchGetResult.rawCells().length); + for (int j = 0; j < getResult.size(); ++j) { + Assert.assertEquals(Bytes.toString(getResult.getRow()), + Bytes.toString(batchGetResult.getRow())); + Assert.assertEquals(Bytes.toString(CellUtil.cloneFamily(getResult.rawCells()[j])), + Bytes.toString(CellUtil.cloneFamily(batchGetResult.rawCells()[j]))); + Assert.assertEquals(Bytes.toString(CellUtil.cloneQualifier(getResult.rawCells()[j])), + Bytes.toString(CellUtil.cloneQualifier(batchGetResult.rawCells()[j]))); + Assert.assertEquals(getResult.rawCells()[j].getTimestamp(), + batchGetResult.rawCells()[j].getTimestamp()); + Assert.assertEquals(Bytes.toString(CellUtil.cloneValue(getResult.rawCells()[j])), + Bytes.toString(CellUtil.cloneValue(batchGetResult.rawCells()[j]))); + } + } + + // multi-cf test + batchLsit.clear(); + for (int i = 0; i < rows; ++i) { + put = new Put(toBytes("Key" + i)); + delete = new Delete(toBytes("Key" + i)); + batchLsit.add(delete); + put.addColumn(family1, family1_column1, family1_value); + put.addColumn(family1, family1_column2, family1_value); + put.addColumn(family1, family1_column3, family1_value); + put.addColumn(family2, family2_column1, family2_value); + put.addColumn(family2, family2_column2, family2_value); + put.addColumn(family3, family3_column1, family3_value); + batchLsit.add(put); + } + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + batchLsit.clear(); + + // get + get1 = new Get("Key1".getBytes()); + get1.addFamily(family1); + get1.addFamily(family2); + batchLsit.add(get1); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + getResult = (Result) results[0]; + for (Cell keyValue : getResult.rawCells()) { + System.out.printf("1. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + Assert.assertFalse(Bytes.toString(CellUtil.cloneFamily(keyValue)).equals("family_with_group3")); + } + // delete + get + batchLsit.clear(); + delete = new Delete(toBytes("Key1")); + batchLsit.add(delete); + batchLsit.add(get1); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(2, results.length); + Assert.assertEquals(0, ((Result) results[1]).rawCells().length); + + // put + get + batchLsit.clear(); + put = new Put(toBytes("Key1")); + put.addColumn(family1, family1_column1, family1_value); + put.addColumn(family2, family2_column1, family2_value); + put.addColumn(family3, family3_column1, family3_value); + batchLsit.add(put); + get1.addFamily(family3); + batchLsit.add(get1); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(2, results.length); + getResult = (Result) results[1]; + for (Cell keyValue : getResult.rawCells()) { + System.out.printf("2. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + + // put + delete + get + // get2 query all family + get2 = new Get("Key2".getBytes()); + batchLsit.clear(); + batchLsit.add(delete); + batchLsit.add(put); + batchLsit.add(get1); + batchLsit.add(delete); + batchLsit.add(get1); + batchLsit.add(get2); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(6, results.length); + key1Result = (Result) results[2]; + Assert.assertEquals(3, key1Result.rawCells().length); + for (Cell keyValue : key1Result.rawCells()) { + System.out.printf("3. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + empResult = (Result) results[4]; + Assert.assertEquals(0, empResult.rawCells().length); + + key2Result = (Result) results[5]; + Assert.assertEquals(6, key2Result.rawCells().length); + for (Cell keyValue : key2Result.rawCells()) { + System.out.printf("4. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + + // set maxVersion + batchLsit.clear(); + put = new Put(toBytes("Key2")); + put.addColumn(family1, family1_column1, family1_value); + put.addColumn(family2, family2_column1, family2_value); + put.addColumn(family3, family3_column1, family3_value); + batchLsit.add(put); + get2.setMaxVersions(10); + batchLsit.add(get2); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(2, results.length); + key2Result = (Result) results[1]; + Assert.assertEquals(9, key2Result.rawCells().length); + for (Cell keyValue : key2Result.rawCells()) { + System.out.printf("5. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + put = new Put(toBytes("Key2")); + put.addColumn(family1, family1_column1, family1_value); + get2.setMaxVersions(2); + get2All = new Get(toBytes("Key2")); + get2All.setMaxVersions(); + batchLsit.clear(); + batchLsit.add(put); + batchLsit.add(get2); + batchLsit.add(get2All); + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(3, results.length); + key2Result = (Result) results[1]; + key2AllResult = (Result) results[2]; + Assert.assertEquals(3, key2AllResult.getColumnCells(family1, family1_column1).size()); + Assert.assertEquals(2, key2Result.getColumnCells(family1, family1_column1).size()); + for (Cell keyValue : key2Result.rawCells()) { + System.out.printf("6. Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n", + Bytes.toString(CellUtil.cloneRow(keyValue)), + Bytes.toString(CellUtil.cloneFamily(keyValue)), + Bytes.toString(CellUtil.cloneQualifier(keyValue)), + keyValue.getTimestamp(), + Bytes.toString(CellUtil.cloneValue(keyValue)) + ); + } + + // compare with get interface + key2GetResult = multiCfHTable.get(get2); + Assert.assertEquals(key2GetResult.rawCells().length, key2Result.rawCells().length); + for (int i = 0; i < key2GetResult.size(); ++i) { + Assert.assertEquals(Bytes.toString(key2Result.getRow()), + Bytes.toString(key2GetResult.getRow())); + Assert.assertEquals(Bytes.toString(CellUtil.cloneFamily(key2Result.rawCells()[i])), + Bytes.toString(CellUtil.cloneFamily(key2GetResult.rawCells()[i]))); + Assert.assertEquals(Bytes.toString(CellUtil.cloneQualifier(key2Result.rawCells()[i])), + Bytes.toString(CellUtil.cloneQualifier(key2GetResult.rawCells()[i]))); + Assert.assertEquals(key2Result.rawCells()[i].getTimestamp(), + key2GetResult.rawCells()[i].getTimestamp()); + Assert.assertEquals(Bytes.toString(CellUtil.cloneValue(key2Result.rawCells()[i])), + Bytes.toString(CellUtil.cloneValue(key2GetResult.rawCells()[i]))); + } + // test batch get + batchLsit.clear(); + for (int i = 0; i < rows; ++i) { + get = new Get(toBytes("Key" + i)); + get.setMaxVersions(10); + batchLsit.add(get); + } + results = new Object[batchLsit.size()]; + multiCfHTable.batch(batchLsit, results); + Assert.assertEquals(rows, results.length); + for (int i = 0; i < rows; ++i) { + get = new Get(toBytes("Key" + i)); + get.setMaxVersions(10); + getResult = multiCfHTable.get(get); + Result batchGetResult = (Result) results[i]; + if (Bytes.toString(get.getRow()).equals("Key1")) { + Assert.assertEquals(0, batchGetResult.rawCells().length); + } else if (Bytes.toString(get.getRow()).equals("Key2")) { + // 6 + 3 + 1 + Assert.assertEquals(10, batchGetResult.rawCells().length); + } else { + Assert.assertEquals(6, batchGetResult.rawCells().length); + } + Assert.assertEquals(getResult.rawCells().length, batchGetResult.rawCells().length); + for (int j = 0; j < getResult.size(); ++j) { + Assert.assertEquals(Bytes.toString(getResult.getRow()), + Bytes.toString(batchGetResult.getRow())); + Assert.assertEquals(Bytes.toString(CellUtil.cloneFamily(getResult.rawCells()[j])), + Bytes.toString(CellUtil.cloneFamily(batchGetResult.rawCells()[j]))); + Assert.assertEquals(Bytes.toString(CellUtil.cloneQualifier(getResult.rawCells()[j])), + Bytes.toString(CellUtil.cloneQualifier(batchGetResult.rawCells()[j]))); + Assert.assertEquals(getResult.rawCells()[j].getTimestamp(), + batchGetResult.rawCells()[j].getTimestamp()); + Assert.assertEquals(Bytes.toString(CellUtil.cloneValue(getResult.rawCells()[j])), + Bytes.toString(CellUtil.cloneValue(batchGetResult.rawCells()[j]))); + } + } } @Test