Repository: scalar-labs/scalardb
Issue: Non-serializable execution (lost update) when a recode is deleted

### Environment
- Version 2.0.0
- with SERIALIZABLE

### How to reproduce
```
CREATE TRANSACTION TABLE foo.foo (
  id TEXT PARTITIONKEY,
  value TEXT
);
```

- Initial state: 

id | value | tx_version (tx metadata) 
-- | ------ | ---
foo |     2 |  1

- Steps
  - T1: start
  - T1: Read a record that has id = foo and put the value into A. If the record does not exist, put 0. 
    - (id, value, tx_version) = ("foo", 2, 1)
  - T2: start
  - T2: Delete a record that has id = foo
  - T2: commit
  - T3: start
  - T3: Read a record that has id = foo and put the value into B. If the record does not exist, put 0. 
  - T3: Add 1 to B and put
    - (id, value, tx_version) = ("foo", 1, 1)
  - T3: commit
  - T1: Add 1 to A and put
    - (id, value, tx_version) = ("foo", 3, 2)
  - T1: commit

- Actual states :

id | value | tx_version
-- | ------ | ------
foo |    3 |     2

The above state can NOT be produced by T1, T2 and T3 in some serial order, which means the execution is not serializable.

- T1->T3->T2 and T3->T1->T2
no record

- T2->T1->T3 and T2->T3->T1

id | value | tx_version
-- | ------ | ------
foo |    2 |     2

- T1->T2->T3 and T3->T2->T1

id | value | tx_version
-- | ------ | ------
foo |   1 |    1


Changed files:
src/main/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposer.java
src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitWithCassandraIntegrationTest.java
src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTest.java
src/test/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposerTest.java

diff --git a/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTest.java b/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTest.java
index 40a9f278..c6dac7ff 100644
--- a/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTest.java
+++ b/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTest.java
@@ -21,6 +21,7 @@ import com.scalar.db.api.Scan;
 import com.scalar.db.api.Selection;
 import com.scalar.db.api.TransactionState;
 import com.scalar.db.exception.storage.ExecutionException;
+import com.scalar.db.exception.storage.NoMutationException;
 import com.scalar.db.exception.transaction.CommitConflictException;
 import com.scalar.db.exception.transaction.CommitException;
 import com.scalar.db.exception.transaction.CoordinatorException;
@@ -1371,6 +1372,93 @@ public class ConsensusCommitIntegrationTest {
     assertThat(thrown2).isInstanceOf(CommitException.class);
   }
 
+  public void putAndCommit_DeleteGivenInBetweenTransactions_ShouldProduceSerializableResults()
+      throws CrudException, CommitException, UnknownTransactionStatusException {
+    // Arrange
+    DistributedTransaction transaction = manager.start();
+    transaction.put(preparePut(0, 0, NAMESPACE, TABLE_1).withValue(new IntValue(BALANCE, 2)));
+    transaction.commit();
+
+    // Act
+    DistributedTransaction transaction1 = manager.start();
+    Optional<Result> result1 = transaction1.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    int balance1 = 0;
+    if (result1.isPresent()) {
+      balance1 = ((IntValue) result1.get().getValue(BALANCE).get()).get();
+    }
+    transaction1.put(
+        preparePut(0, 0, NAMESPACE, TABLE_1).withValue(new IntValue(BALANCE, balance1 + 1)));
+
+    DistributedTransaction transaction2 = manager.start();
+    transaction2.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    transaction2.delete(prepareDelete(0, 0, NAMESPACE, TABLE_1));
+    transaction2.commit();
+
+    // the same transaction processing as transaction1
+    DistributedTransaction transaction3 = manager.start();
+    Optional<Result> result3 = transaction3.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    int balance3 = 0;
+    if (result3.isPresent()) {
+      balance3 = ((IntValue) result3.get().getValue(BALANCE).get()).get();
+    }
+    transaction3.put(
+        preparePut(0, 0, NAMESPACE, TABLE_1).withValue(new IntValue(BALANCE, balance3 + 1)));
+    transaction3.commit();
+
+    Throwable thrown = catchThrowable(() -> transaction1.commit());
+
+    // Assert
+    assertThat(thrown)
+        .isInstanceOf(CommitConflictException.class)
+        .hasCauseInstanceOf(NoMutationException.class);
+    transaction = manager.start();
+    Optional<Result> result = transaction.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    assertThat(((IntValue) result.get().getValue(BALANCE).get()).get()).isEqualTo(1);
+  }
+
+  public void deleteAndCommit_DeleteGivenInBetweenTransactions_ShouldProduceSerializableResults()
+      throws CrudException, CommitException, UnknownTransactionStatusException {
+    // Arrange
+    DistributedTransaction transaction = manager.start();
+    transaction.put(preparePut(0, 0, NAMESPACE, TABLE_1).withValue(new IntValue(BALANCE, 2)));
+    transaction.commit();
+
+    // Act
+    DistributedTransaction transaction1 = manager.start();
+    Optional<Result> result1 = transaction1.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    int balance1 = 0;
+    if (result1.isPresent()) {
+      balance1 = ((IntValue) result1.get().getValue(BALANCE).get()).get();
+    }
+    transaction1.delete(prepareDelete(0, 0, NAMESPACE, TABLE_1));
+
+    DistributedTransaction transaction2 = manager.start();
+    transaction2.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    transaction2.delete(prepareDelete(0, 0, NAMESPACE, TABLE_1));
+    transaction2.commit();
+
+    // the same transaction processing as transaction1
+    DistributedTransaction transaction3 = manager.start();
+    Optional<Result> result3 = transaction3.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    int balance3 = 0;
+    if (result3.isPresent()) {
+      balance3 = ((IntValue) result3.get().getValue(BALANCE).get()).get();
+    }
+    transaction3.put(
+        preparePut(0, 0, NAMESPACE, TABLE_1).withValue(new IntValue(BALANCE, balance3 + 1)));
+    transaction3.commit();
+
+    Throwable thrown = catchThrowable(() -> transaction1.commit());
+
+    // Assert
+    assertThat(thrown)
+        .isInstanceOf(CommitConflictException.class)
+        .hasCauseInstanceOf(NoMutationException.class);
+    transaction = manager.start();
+    Optional<Result> result = transaction.get(prepareGet(0, 0, NAMESPACE, TABLE_1));
+    assertThat(((IntValue) result.get().getValue(BALANCE).get()).get()).isEqualTo(1);
+  }
+
   private ConsensusCommit prepareTransfer(int fromId, int toId, int amount) throws CrudException {
     ConsensusCommit transaction = manager.start();
     List<Get> gets = prepareGets(NAMESPACE, TABLE_1);
diff --git a/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitWithCassandraIntegrationTest.java b/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitWithCassandraIntegrationTest.java
index db18b2d3..59c2ce77 100644
--- a/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitWithCassandraIntegrationTest.java
+++ b/src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitWithCassandraIntegrationTest.java
@@ -359,6 +359,18 @@ public class ConsensusCommitWithCassandraIntegrationTest {
     test.commit_WriteSkewWithScanOnNonExistingRecordsWithSerializable_ShouldThrowCommitException();
   }
 
+  @Test
+  public void putAndCommit_DeleteGivenInBetweenTransactions_ShouldProduceSerializableResults()
+      throws Exception {
+    test.putAndCommit_DeleteGivenInBetweenTransactions_ShouldProduceSerializableResults();
+  }
+
+  @Test
+  public void deleteAndCommit_DeleteGivenInBetweenTransactions_ShouldProduceSerializableResults()
+      throws Exception {
+    test.deleteAndCommit_DeleteGivenInBetweenTransactions_ShouldProduceSerializableResults();
+  }
+
   @BeforeClass
   public static void setUpBeforeClass() throws IOException, InterruptedException {
     executeStatement(createNamespaceStatement(ConsensusCommitIntegrationTest.NAMESPACE));
diff --git a/src/main/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposer.java b/src/main/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposer.java
index 247880de..979b66d6 100644
--- a/src/main/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposer.java
+++ b/src/main/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposer.java
@@ -1,7 +1,10 @@
 package com.scalar.db.transaction.consensuscommit;
 
 import static com.scalar.db.api.ConditionalExpression.Operator;
-import static com.scalar.db.transaction.consensuscommit.Attribute.*;
+import static com.scalar.db.transaction.consensuscommit.Attribute.ID;
+import static com.scalar.db.transaction.consensuscommit.Attribute.VERSION;
+import static com.scalar.db.transaction.consensuscommit.Attribute.toIdValue;
+import static com.scalar.db.transaction.consensuscommit.Attribute.toVersionValue;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.scalar.db.api.ConditionalExpression;
@@ -64,7 +67,9 @@ public class PrepareMutationComposer extends AbstractMutationComposer {
 
       // check if the record is not interrupted by other conflicting transactions
       put.withCondition(
-          new PutIf(new ConditionalExpression(VERSION, toVersionValue(version), Operator.EQ)));
+          new PutIf(
+              new ConditionalExpression(VERSION, toVersionValue(version), Operator.EQ),
+              new ConditionalExpression(ID, toIdValue(result.getId()), Operator.EQ)));
     } else { // initial record
       values.add(Attribute.toVersionValue(1));
 
@@ -98,7 +103,9 @@ public class PrepareMutationComposer extends AbstractMutationComposer {
 
     // check if the record is not interrupted by other conflicting transactions
     put.withCondition(
-        new PutIf(new ConditionalExpression(VERSION, toVersionValue(version), Operator.EQ)));
+        new PutIf(
+            new ConditionalExpression(VERSION, toVersionValue(version), Operator.EQ),
+            new ConditionalExpression(ID, toIdValue(result.getId()), Operator.EQ)));
 
     put.withValues(values);
     mutations.add(put);
diff --git a/src/test/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposerTest.java b/src/test/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposerTest.java
index 0096a06b..0e2f59d5 100644
--- a/src/test/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposerTest.java
+++ b/src/test/java/com/scalar/db/transaction/consensuscommit/PrepareMutationComposerTest.java
@@ -1,7 +1,10 @@
 package com.scalar.db.transaction.consensuscommit;
 
 import static com.scalar.db.api.ConditionalExpression.Operator;
-import static com.scalar.db.transaction.consensuscommit.Attribute.*;
+import static com.scalar.db.transaction.consensuscommit.Attribute.ID;
+import static com.scalar.db.transaction.consensuscommit.Attribute.VERSION;
+import static com.scalar.db.transaction.consensuscommit.Attribute.toIdValue;
+import static com.scalar.db.transaction.consensuscommit.Attribute.toVersionValue;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
@@ -120,7 +123,9 @@ public class PrepareMutationComposerTest {
     Put actual = (Put) mutations.get(0);
     put.withConsistency(Consistency.LINEARIZABLE);
     put.withCondition(
-        new PutIf(new ConditionalExpression(VERSION, toVersionValue(2), Operator.EQ)));
+        new PutIf(
+            new ConditionalExpression(VERSION, toVersionValue(2), Operator.EQ),
+            new ConditionalExpression(ID, toIdValue(ANY_ID_2), Operator.EQ)));
     put.withValue(Attribute.toPreparedAtValue(ANY_TIME_5));
     put.withValue(Attribute.toIdValue(ANY_ID_3));
     put.withValue(Attribute.toStateValue(TransactionState.PREPARED));
@@ -170,7 +175,9 @@ public class PrepareMutationComposerTest {
             .forTable(delete.forTable().get());
     expected.withConsistency(Consistency.LINEARIZABLE);
     expected.withCondition(
-        new PutIf(new ConditionalExpression(VERSION, toVersionValue(2), Operator.EQ)));
+        new PutIf(
+            new ConditionalExpression(VERSION, toVersionValue(2), Operator.EQ),
+            new ConditionalExpression(ID, toIdValue(ANY_ID_2), Operator.EQ)));
     expected.withValue(Attribute.toPreparedAtValue(ANY_TIME_5));
     expected.withValue(Attribute.toIdValue(ANY_ID_3));
     expected.withValue(Attribute.toStateValue(TransactionState.DELETED));