From b97c3daf9066afca903b695a296881ab77d23598 Mon Sep 17 00:00:00 2001
From: Enis Soztutar <enis@apache.org>
Date: Tue, 18 Nov 2014 11:34:49 -0800
Subject: [PATCH] HBASE-12449 Use the max timestamp of current or old cell's
 timestamp in HRegion.append()

---
 .../org/apache/hadoop/hbase/regionserver/HRegion.java | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index fbf151af9b6..d7f83c9ce76 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -768,7 +768,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
       // In distributedLogReplay mode, we don't know the last change sequence number because region
       // is opened before recovery completes. So we add a safety bumper to avoid new sequence number
       // overlaps used sequence numbers
-      nextSeqid = HLogUtil.writeRegionOpenSequenceIdFile(this.fs.getFileSystem(), 
+      nextSeqid = HLogUtil.writeRegionOpenSequenceIdFile(this.fs.getFileSystem(),
             this.fs.getRegionDir(), nextSeqid, (this.flushPerChanges + 10000000));
     }
 
@@ -3190,7 +3190,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
       Store store = getStore(family);
       int listSize = cells.size();
       for (int i=0; i < listSize; i++) {
-        Cell cell = cells.get(i);  
+        Cell cell = cells.get(i);
         CellUtil.setSequenceId(cell, mvccNum);
         Pair<Long, Cell> ret = store.add(cell);
         size += ret.getFirst();
@@ -5390,9 +5390,10 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
               if (idx < results.size()
                   && CellUtil.matchingQualifier(results.get(idx), cell)) {
                 oldCell = results.get(idx);
+                long ts = Math.max(now, oldCell.getTimestamp());
                 // allocate an empty kv once
                 newCell = new KeyValue(row.length, cell.getFamilyLength(),
-                    cell.getQualifierLength(), now, KeyValue.Type.Put,
+                    cell.getQualifierLength(), ts, KeyValue.Type.Put,
                     oldCell.getValueLength() + cell.getValueLength(),
                     oldCell.getTagsLength() + cell.getTagsLength());
                 // copy in the value
@@ -5603,8 +5604,10 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
               boolean noWriteBack = (amount == 0);
 
               Cell c = null;
+              long ts = now;
               if (idx < results.size() && CellUtil.matchingQualifier(results.get(idx), kv)) {
                 c = results.get(idx);
+                ts = Math.max(now, c.getTimestamp());
                 if(c.getValueLength() == Bytes.SIZEOF_LONG) {
                   amount += Bytes.toLong(c.getValueArray(), c.getValueOffset(), Bytes.SIZEOF_LONG);
                 } else {
@@ -5620,7 +5623,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver { //
               byte[] val = Bytes.toBytes(amount);
               int oldCellTagsLen = (c == null) ? 0 : c.getTagsLength();
               int incCellTagsLen = kv.getTagsLength();
-              Cell newKV = new KeyValue(row.length, family.getKey().length, q.length, now,
+              Cell newKV = new KeyValue(row.length, family.getKey().length, q.length, ts,
                   KeyValue.Type.Put, val.length, oldCellTagsLen + incCellTagsLen);
               System.arraycopy(row, 0, newKV.getRowArray(), newKV.getRowOffset(), row.length);
               System.arraycopy(family.getKey(), 0, newKV.getFamilyArray(), newKV.getFamilyOffset(),