Skip to content

Commit

Permalink
HBASE-26713 Default to LATEST_TIMESTAMP if no timestamp sent along on…
Browse files Browse the repository at this point in the history 
… Increment/Append (#4075)

Signed-off-by: Andrew Purtell <[email protected]>
Signed-off-by: Viraj Jasani <[email protected]>
  • Loading branch information
@bbeaudreault @apurtell
bbeaudreault authored and apurtell committed Jan 28, 2022
1 parent 70018b0 commit 590e38e
Show file tree
Hide file tree
Showing 4 changed files with 164 additions and 72 deletions.
15 changes: 10 additions & 5 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -625,10 +625,7 @@ public static Delete toDelete(final MutationProto proto, final CellScanner cellS
if (qv.hasQualifier()) {
qualifier = qv.getQualifier().toByteArray();
}
long ts = HConstants.LATEST_TIMESTAMP;
if (qv.hasTimestamp()) {
ts = qv.getTimestamp();
}
long ts = cellTimestampOrLatest(qv);
if (deleteType == DeleteType.DELETE_ONE_VERSION) {
delete.addColumn(family, qualifier, ts);
} else if (deleteType == DeleteType.DELETE_MULTIPLE_VERSIONS) {
Expand Down Expand Up @@ -692,7 +689,7 @@ private static <T extends Mutation> T toDelta(Function<Bytes, T> supplier, Consu
if (qv.hasTags()) {
tags = qv.getTags().toByteArray();
}
consumer.accept(mutation, CellUtil.createCell(mutation.getRow(), family, qualifier, qv.getTimestamp(),
consumer.accept(mutation, CellUtil.createCell(mutation.getRow(), family, qualifier, cellTimestampOrLatest(qv),
KeyValue.Type.Put, value, tags));
}
}
Expand All @@ -704,6 +701,14 @@ private static <T extends Mutation> T toDelta(Function<Bytes, T> supplier, Consu
return mutation;
}

private static long cellTimestampOrLatest(QualifierValue cell) {
if (cell.hasTimestamp()) {
return cell.getTimestamp();
} else {
return HConstants.LATEST_TIMESTAMP;
}
}

/**
* Convert a protocol buffer Mutate to an Append
* @param cellScanner
Expand Down
15 changes: 10 additions & 5 deletions hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -785,10 +785,7 @@ public static Delete toDelete(final MutationProto proto, final CellScanner cellS
if (qv.hasQualifier()) {
qualifier = qv.getQualifier().toByteArray();
}
long ts = HConstants.LATEST_TIMESTAMP;
if (qv.hasTimestamp()) {
ts = qv.getTimestamp();
}
long ts = cellTimestampOrLatest(qv);
if (deleteType == DeleteType.DELETE_ONE_VERSION) {
delete.addColumn(family, qualifier, ts);
} else if (deleteType == DeleteType.DELETE_MULTIPLE_VERSIONS) {
Expand Down Expand Up @@ -855,7 +852,7 @@ private static <T extends Mutation> T toDelta(Function<Bytes, T> supplier, Consu
.setRow(mutation.getRow())
.setFamily(family)
.setQualifier(qualifier)
.setTimestamp(qv.getTimestamp())
.setTimestamp(cellTimestampOrLatest(qv))
.setType(KeyValue.Type.Put.getCode())
.setValue(value)
.setTags(tags)
Expand All @@ -870,6 +867,14 @@ private static <T extends Mutation> T toDelta(Function<Bytes, T> supplier, Consu
return mutation;
}

private static long cellTimestampOrLatest(QualifierValue cell) {
if (cell.hasTimestamp()) {
return cell.getTimestamp();
} else {
return HConstants.LATEST_TIMESTAMP;
}
}

/**
* Convert a protocol buffer Mutate to an Append
* @param cellScanner
Expand Down
103 changes: 73 additions & 30 deletions hbase-client/src/test/java/org/apache/hadoop/hbase/shaded/protobuf/TestProtobufUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.hadoop.hbase.ExtendedCellBuilder;
import org.apache.hadoop.hbase.ExtendedCellBuilderFactory;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.PrivateCellUtil;
import org.apache.hadoop.hbase.Tag;
Expand Down Expand Up @@ -294,7 +295,22 @@ public void testToCell() {
*/
@Test
public void testIncrement() throws IOException {
long timeStamp = 111111;

MutationProto proto = getIncrementMutation(111111L);
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
MutationProto.Builder mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Increment increment = ProtobufUtil.toIncrement(proto, null);
mutateBuilder.setTimestamp(increment.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(increment.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.INCREMENT, increment));
}

private MutationProto getIncrementMutation(Long timestamp) {
MutationProto.Builder mutateBuilder = MutationProto.newBuilder();
mutateBuilder.setRow(ByteString.copyFromUtf8("row"));
mutateBuilder.setMutateType(MutationProto.MutationType.INCREMENT);
Expand All @@ -303,66 +319,93 @@ public void testIncrement() throws IOException {
QualifierValue.Builder qualifierBuilder = QualifierValue.newBuilder();
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c1"));
qualifierBuilder.setValue(ByteString.copyFrom(Bytes.toBytes(11L)));
qualifierBuilder.setTimestamp(timeStamp);

if (timestamp != null) {
qualifierBuilder.setTimestamp(timestamp);
}

valueBuilder.addQualifierValue(qualifierBuilder.build());
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c2"));
qualifierBuilder.setValue(ByteString.copyFrom(Bytes.toBytes(22L)));
valueBuilder.addQualifierValue(qualifierBuilder.build());
mutateBuilder.addColumnValue(valueBuilder.build());

MutationProto proto = mutateBuilder.build();
return mutateBuilder.build();
}

/**
* Older clients may not send along a timestamp in the MutationProto. Check that we
* default correctly.
*/
@Test
public void testIncrementNoTimestamp() throws IOException {
MutationProto mutation = getIncrementMutation(null);
Increment increment = ProtobufUtil.toIncrement(mutation, null);
assertEquals(HConstants.LATEST_TIMESTAMP, increment.getTimestamp());
increment.getFamilyCellMap().values()
.forEach(cells ->
cells.forEach(cell ->
assertEquals(HConstants.LATEST_TIMESTAMP, cell.getTimestamp())));
}

/**
* Test Append Mutate conversions.
*
* @throws IOException if converting to an {@link Append} fails
*/
@Test
public void testAppend() throws IOException {
MutationProto proto = getAppendMutation(111111L);
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
mutateBuilder = MutationProto.newBuilder(proto);
MutationProto.Builder mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Increment increment = ProtobufUtil.toIncrement(proto, null);
mutateBuilder.setTimestamp(increment.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(increment.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.INCREMENT, increment));
Append append = ProtobufUtil.toAppend(proto, null);

// append always use the latest timestamp,
// reset the timestamp to the original mutate
mutateBuilder.setTimestamp(append.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(append.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.APPEND, append));
}

/**
* Test Append Mutate conversions.
*
* @throws IOException if converting to an {@link Append} fails
* Older clients may not send along a timestamp in the MutationProto. Check that we
* default correctly.
*/
@Test
public void testAppend() throws IOException {
long timeStamp = 111111;
public void testAppendNoTimestamp() throws IOException {
MutationProto mutation = getAppendMutation(null);
Append append = ProtobufUtil.toAppend(mutation, null);
assertEquals(HConstants.LATEST_TIMESTAMP, append.getTimestamp());
append.getFamilyCellMap().values().forEach(cells -> cells.forEach(cell -> assertEquals(HConstants.LATEST_TIMESTAMP, cell.getTimestamp())));
}

private MutationProto getAppendMutation(Long timestamp) {
MutationProto.Builder mutateBuilder = MutationProto.newBuilder();
mutateBuilder.setRow(ByteString.copyFromUtf8("row"));
mutateBuilder.setMutateType(MutationType.APPEND);
mutateBuilder.setTimestamp(timeStamp);
if (timestamp != null) {
mutateBuilder.setTimestamp(timestamp);
}
ColumnValue.Builder valueBuilder = ColumnValue.newBuilder();
valueBuilder.setFamily(ByteString.copyFromUtf8("f1"));
QualifierValue.Builder qualifierBuilder = QualifierValue.newBuilder();
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c1"));
qualifierBuilder.setValue(ByteString.copyFromUtf8("v1"));
qualifierBuilder.setTimestamp(timeStamp);
if (timestamp != null) {
qualifierBuilder.setTimestamp(timestamp);
}
valueBuilder.addQualifierValue(qualifierBuilder.build());
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c2"));
qualifierBuilder.setValue(ByteString.copyFromUtf8("v2"));
valueBuilder.addQualifierValue(qualifierBuilder.build());
mutateBuilder.addColumnValue(valueBuilder.build());

MutationProto proto = mutateBuilder.build();
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Append append = ProtobufUtil.toAppend(proto, null);

// append always use the latest timestamp,
// reset the timestamp to the original mutate
mutateBuilder.setTimestamp(append.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(append.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.APPEND, append));
return mutateBuilder.build();
}

private static ProcedureProtos.Procedure.Builder createProcedureBuilder(long procId) {
Expand Down
103 changes: 71 additions & 32 deletions hbase-server/src/test/java/org/apache/hadoop/hbase/protobuf/TestProtobufUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,38 +122,57 @@ public void testGet() throws IOException {
*/
@Test
public void testAppend() throws IOException {
long timeStamp = 111111;
MutationProto proto = getAppendMutation(111111L);
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
MutationProto.Builder mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Append append = ProtobufUtil.toAppend(proto, null);

// append always use the latest timestamp,
// reset the timestamp to the original mutate
mutateBuilder.setTimestamp(append.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(append.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.APPEND, append));
}

/**
* Older clients may not send along a timestamp in the MutationProto. Check that we
* default correctly.
*/
@Test
public void testAppendNoTimestamp() throws IOException {
MutationProto mutation = getAppendMutation(null);
Append append = ProtobufUtil.toAppend(mutation, null);
assertEquals(HConstants.LATEST_TIMESTAMP, append.getTimestamp());
append.getFamilyCellMap().values().forEach(cells -> cells.forEach(cell -> assertEquals(HConstants.LATEST_TIMESTAMP, cell.getTimestamp())));
}

private MutationProto getAppendMutation(Long timestamp) {
MutationProto.Builder mutateBuilder = MutationProto.newBuilder();
mutateBuilder.setRow(ByteString.copyFromUtf8("row"));
mutateBuilder.setMutateType(MutationType.APPEND);
mutateBuilder.setTimestamp(timeStamp);
if (timestamp != null) {
mutateBuilder.setTimestamp(timestamp);
}
ColumnValue.Builder valueBuilder = ColumnValue.newBuilder();
valueBuilder.setFamily(ByteString.copyFromUtf8("f1"));
QualifierValue.Builder qualifierBuilder = QualifierValue.newBuilder();
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c1"));
qualifierBuilder.setValue(ByteString.copyFromUtf8("v1"));
qualifierBuilder.setTimestamp(timeStamp);
if (timestamp != null) {
qualifierBuilder.setTimestamp(timestamp);
}
valueBuilder.addQualifierValue(qualifierBuilder.build());
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c2"));
qualifierBuilder.setValue(ByteString.copyFromUtf8("v2"));
valueBuilder.addQualifierValue(qualifierBuilder.build());
mutateBuilder.addColumnValue(valueBuilder.build());

MutationProto proto = mutateBuilder.build();
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Append append = ProtobufUtil.toAppend(proto, null);

// append always use the latest timestamp,
// reset the timestamp to the original mutate
mutateBuilder.setTimestamp(append.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(append.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.APPEND, append));
return mutateBuilder.build();
}

/**
Expand Down Expand Up @@ -210,7 +229,36 @@ public void testDelete() throws IOException {
*/
@Test
public void testIncrement() throws IOException {
long timeStamp = 111111;
MutationProto proto = getIncrementMutation(111111L);
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
MutationProto.Builder mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Increment increment = ProtobufUtil.toIncrement(proto, null);
mutateBuilder.setTimestamp(increment.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(increment.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.INCREMENT, increment));
}

/**
* Older clients may not send along a timestamp in the MutationProto. Check that we
* default correctly.
*/
@Test
public void testIncrementNoTimestamp() throws IOException {
MutationProto mutation = getIncrementMutation(null);
Increment increment = ProtobufUtil.toIncrement(mutation, null);
assertEquals(HConstants.LATEST_TIMESTAMP, increment.getTimestamp());
increment.getFamilyCellMap().values()
.forEach(cells ->
cells.forEach(cell ->
assertEquals(HConstants.LATEST_TIMESTAMP, cell.getTimestamp())));
}

private MutationProto getIncrementMutation(Long timestamp) {
MutationProto.Builder mutateBuilder = MutationProto.newBuilder();
mutateBuilder.setRow(ByteString.copyFromUtf8("row"));
mutateBuilder.setMutateType(MutationType.INCREMENT);
Expand All @@ -219,25 +267,16 @@ public void testIncrement() throws IOException {
QualifierValue.Builder qualifierBuilder = QualifierValue.newBuilder();
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c1"));
qualifierBuilder.setValue(ByteString.copyFrom(Bytes.toBytes(11L)));
qualifierBuilder.setTimestamp(timeStamp);
if (timestamp != null) {
qualifierBuilder.setTimestamp(timestamp);
}
valueBuilder.addQualifierValue(qualifierBuilder.build());
qualifierBuilder.setQualifier(ByteString.copyFromUtf8("c2"));
qualifierBuilder.setValue(ByteString.copyFrom(Bytes.toBytes(22L)));
valueBuilder.addQualifierValue(qualifierBuilder.build());
mutateBuilder.addColumnValue(valueBuilder.build());

MutationProto proto = mutateBuilder.build();
// default fields
assertEquals(MutationProto.Durability.USE_DEFAULT, proto.getDurability());

// set the default value for equal comparison
mutateBuilder = MutationProto.newBuilder(proto);
mutateBuilder.setDurability(MutationProto.Durability.USE_DEFAULT);

Increment increment = ProtobufUtil.toIncrement(proto, null);
mutateBuilder.setTimestamp(increment.getTimestamp());
mutateBuilder.setTimeRange(ProtobufUtil.toTimeRange(increment.getTimeRange()));
assertEquals(mutateBuilder.build(), ProtobufUtil.toMutation(MutationType.INCREMENT, increment));
return mutateBuilder.build();
}

/**
Expand Down

0 comments on commit 590e38e

Please sign in to comment.