IncrementAdapter implements OperationAdapter<Increment, ReadModifyWriteRow> { @Override public void adapt(Increment operation, ReadModifyWriteRow readModifyWriteRow) { if (!operation.getTimeRange().isAllTime()) { throw new UnsupportedOperationException( "Setting the time range in an Increment is not implemented"); } for (Map.Entry<byte[], NavigableMap<byte[], Long>> familyEntry : operation.getFamilyMapOfLongs().entrySet()) { String familyName = Bytes.toString(familyEntry.getKey()); List<Cell> mutationCells = CellDeduplicationHelper.deduplicateFamily(operation, familyEntry.getKey()); for (Cell cell : mutationCells) { readModifyWriteRow.increment( familyName, ByteString.copyFrom( cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()), Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())); } } }  @Override void adapt(Increment operation, ReadModifyWriteRow readModifyWriteRow);  }
@Test public void testBasicRowKeyIncrement() { byte[] rowKey = dataHelper.randomData("rk1-"); Increment incr = new Increment(rowKey); ReadModifyWriteRow readModifyWriteRow = ReadModifyWriteRow.create(TABLE_ID, ByteString.copyFrom(rowKey)); incrementAdapter.adapt(incr, readModifyWriteRow); ReadModifyWriteRowRequest requestBuilder = readModifyWriteRow.toProto(requestContext); ByteString adaptedRowKey = requestBuilder.getRowKey(); Assert.assertArrayEquals(rowKey, adaptedRowKey.toByteArray()); }
@Test public void testSingleIncrement() { byte[] rowKey = dataHelper.randomData("rk1-"); byte[] family = Bytes.toBytes("family"); byte[] qualifier = Bytes.toBytes("qualifier"); long amount = 1234; Increment incr = new Increment(rowKey); incr.addColumn(family, qualifier, amount); ReadModifyWriteRow readModifyWriteRow = ReadModifyWriteRow.create(TABLE_ID, ByteString.copyFrom(rowKey)); incrementAdapter.adapt(incr, readModifyWriteRow); ReadModifyWriteRowRequest requestBuilder = readModifyWriteRow.toProto(requestContext); Assert.assertEquals(1, requestBuilder.getRulesCount()); ReadModifyWriteRule rule = requestBuilder.getRules(0); Assert.assertEquals("qualifier", rule.getColumnQualifier().toStringUtf8()); Assert.assertEquals("family", rule.getFamilyName()); Assert.assertEquals(amount, rule.getIncrementAmount()); }
@Test public void testMultipleIncrement() { byte[] rowKey = dataHelper.randomData("rk1-"); byte[] family1 = Bytes.toBytes("family1"); byte[] qualifier1 = Bytes.toBytes("qualifier1"); long amount1 = 1234; byte[] family2 = Bytes.toBytes("family2"); byte[] qualifier2 = Bytes.toBytes("qualifier2"); long amount2 = 4321; Increment incr = new Increment(rowKey); incr.addColumn(family1, qualifier1, amount1); incr.addColumn(family2, qualifier2, amount2); ReadModifyWriteRow readModifyWriteRow = ReadModifyWriteRow.create(TABLE_ID, ByteString.copyFrom(rowKey)); incrementAdapter.adapt(incr, readModifyWriteRow); ReadModifyWriteRowRequest requestBuilder = readModifyWriteRow.toProto(requestContext); Assert.assertEquals(2, requestBuilder.getRulesCount()); ReadModifyWriteRule rule = requestBuilder.getRules(0); Assert.assertEquals("family1", rule.getFamilyName()); Assert.assertEquals("qualifier1", rule.getColumnQualifier().toStringUtf8()); Assert.assertEquals(amount1, rule.getIncrementAmount()); rule = requestBuilder.getRules(1); Assert.assertEquals("family2", rule.getFamilyName()); Assert.assertEquals("qualifier2", rule.getColumnQualifier().toStringUtf8()); Assert.assertEquals(amount2, rule.getIncrementAmount()); }
@Test public void testMultipleIncrementWithDuplicateQualifier() { byte[] rowKey = dataHelper.randomData("rk1-"); byte[] family1 = Bytes.toBytes("family1"); byte[] qualifier1 = Bytes.toBytes("qualifier1"); long amount1 = 1234; byte[] family2 = Bytes.toBytes("family2"); byte[] qualifier2 = Bytes.toBytes("qualifier2"); long amount2 = 4321; long amount3 = 5000; Increment incr = new Increment(rowKey); incr.addColumn(family1, qualifier1, amount1); incr.addColumn(family2, qualifier2, amount2); incr.addColumn(family2, qualifier2, amount3); ReadModifyWriteRow readModifyWriteRow = ReadModifyWriteRow.create(TABLE_ID, ByteString.copyFrom(rowKey)); incrementAdapter.adapt(incr, readModifyWriteRow); ReadModifyWriteRowRequest requestBuilder = readModifyWriteRow.toProto(requestContext); Assert.assertEquals(2, requestBuilder.getRulesCount()); ReadModifyWriteRule rule = requestBuilder.getRules(0); Assert.assertEquals("family1", rule.getFamilyName()); Assert.assertEquals("qualifier1", rule.getColumnQualifier().toStringUtf8()); Assert.assertEquals(amount1, rule.getIncrementAmount()); rule = requestBuilder.getRules(1); Assert.assertEquals("family2", rule.getFamilyName()); Assert.assertEquals("qualifier2", rule.getColumnQualifier().toStringUtf8()); Assert.assertEquals(amount3, rule.getIncrementAmount()); }
@Test public void testIncrementTimeRange() throws IOException { byte[] rowKey = dataHelper.randomData("rk1-"); Increment incr = new Increment(rowKey); incr.setTimeRange(0, 10); expectedException.expect(UnsupportedOperationException.class); expectedException.expectMessage("Setting the time range in an Increment is not implemented"); ReadModifyWriteRow readModifyWriteRow = ReadModifyWriteRow.create(TABLE_ID, ByteString.copyFrom(rowKey)); incrementAdapter.adapt(incr, readModifyWriteRow); }