txn: add CheckNotExist mutation when delete-your-writes happens

Signed-off-by: Ziyi Yan <ziyi.yan@foxmail.com>

Author: ziyi-yan

src/transaction/buffer.rs

@@ -211,7 +211,20 @@ impl Buffer {
 
     /// Mark a value as deleted.
     pub async fn delete(&self, key: Key) {
-        self.mutations.lock().await.insert(key, BufferEntry::Del);
+        let mut mutations = self.mutations.lock().await;
+        let value = mutations
+            .entry_map
+            .entry(key.clone())
+            .or_insert(BufferEntry::Del);
+
+        let new_value: BufferEntry;
+        if let BufferEntry::Insert(_) = value {
+            new_value = BufferEntry::CheckNotExist
+        } else {
+            new_value = BufferEntry::Del
+        }
+
+        mutations.insert(key, new_value);
     }
 
     /// Converts the buffered mutations to the proto buffer version
@@ -257,6 +270,9 @@ impl Buffer {
             Some(BufferEntry::Insert(v)) => {
                 assert!(value.as_ref() == Some(v))
             }
+            Some(BufferEntry::CheckNotExist) => {
+                assert!(value.is_none());
+            }
         }
     }
 }
@@ -268,6 +284,7 @@ impl Buffer {
 //   - `Put`
 //   - `Del`
 //   - `Insert`
+//   - `CheckNotExist`, a constraint to ensure the key doesn't exist. See https://github.com/pingcap/tidb/pull/14968.
 // Cache of read requests:
 //   - `Cached`, generated by normal read requests
 //   - `ReadLockCached`, generated by lock commands (`lock_keys`, `get_for_update`) and optionally read requests
@@ -297,6 +314,8 @@ enum BufferEntry {
     Del,
     // Key should be check not exists before.
     Insert(Value),
+    // Key should be check not exists before.
+    CheckNotExist,
 }
 
 impl BufferEntry {
@@ -314,6 +333,7 @@ impl BufferEntry {
                 pb.set_op(kvrpcpb::Op::Insert);
                 pb.set_value(v.clone());
             }
+            BufferEntry::CheckNotExist => pb.set_op(kvrpcpb::Op::CheckNotExists),
         };
         pb.set_key(key.clone().into());
         Some(pb)
@@ -327,6 +347,7 @@ impl BufferEntry {
             BufferEntry::Locked(None) => MutationValue::Undetermined,
             BufferEntry::Locked(Some(value)) => MutationValue::Determined(value.clone()),
             BufferEntry::Insert(value) => MutationValue::Determined(Some(value.clone())),
+            BufferEntry::CheckNotExist => MutationValue::Determined(None),
         }
     }
 }