Merge pull request #135 from scalar-labs/bugfix-inconsistent-read

Fix inconsistent read bug

src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTest.java

@@ -114,11 +114,30 @@ public void get_CalledTwice_ShouldReturnFromSnapshotInSecondTime()
     Get get = prepareGet(0, 0, NAMESPACE, TABLE_1);
 
     // Act
-    transaction.get(get);
-    transaction.get(get);
+    Optional<Result> result1 = transaction.get(get);
+    Optional<Result> result2 = transaction.get(get);
 
     // Assert
     verify(storage).get(any(Get.class));
+    assertThat(result1).isEqualTo(result2);
+  }
+
+  public void
+      get_CalledTwiceAndAnotherTransactionCommitsInBetween_ShouldReturnFromSnapshotInSecondTime()
+          throws CrudException, ExecutionException, CommitException,
+              UnknownTransactionStatusException {
+    // Arrange
+    transaction = manager.start();
+    Get get = prepareGet(0, 0, NAMESPACE, TABLE_1);
+
+    // Act
+    Optional<Result> result1 = transaction.get(get);
+    populateRecords();
+    Optional<Result> result2 = transaction.get(get);
+
+    // Assert
+    verify(storage).get(any(Get.class));
+    assertThat(result1).isEqualTo(result2);
   }
 
   public void get_GetGivenForNonExisting_ShouldReturnEmpty()

src/integration-test/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitWithCassandraIntegrationTest.java

@@ -72,6 +72,14 @@ public void get_CalledTwice_ShouldReturnFromSnapshotInSecondTime() throws Except
     test.get_CalledTwice_ShouldReturnFromSnapshotInSecondTime();
   }
 
+  @Test
+  public void
+      get_CalledTwiceAndAnotherTransactionCommitsInBetween_ShouldReturnFromSnapshotInSecondTime()
+          throws Exception {
+    test
+        .get_CalledTwiceAndAnotherTransactionCommitsInBetween_ShouldReturnFromSnapshotInSecondTime();
+  }
+
   @Test
   public void get_GetGivenForNonExisting_ShouldReturnEmpty() throws Exception {
     test.get_GetGivenForNonExisting_ShouldReturnEmpty();

src/main/java/com/scalar/db/transaction/consensuscommit/CrudHandler.java

@@ -39,9 +39,8 @@ public Optional<Result> get(Get get) throws CrudException {
     Optional<TransactionResult> result;
     Snapshot.Key key = new Snapshot.Key(get);
 
-    result = snapshot.get(key);
-    if (result.isPresent()) {
-      return Optional.of(result.get());
+    if (snapshot.containsKey(key)) {
+      return snapshot.get(key).map(r -> r);
     }
 
     result = getFromStorage(get);
@@ -80,11 +79,11 @@ public List<Result> scan(Scan scan) throws CrudException {
             getSnapshotKey(r, scan)
                 .orElseThrow(() -> new CrudRuntimeException("can't get a snapshot key"));
 
-        if (snapshot.get(key).isPresent()) {
-          result = snapshot.get(key).get();
+        if (snapshot.containsKey(key)) {
+          result = snapshot.get(key).orElse(null);
         }
 
-        snapshot.put(key, Optional.of(result));
+        snapshot.put(key, Optional.ofNullable(result));
         keys.add(key);
         results.add(result);
       }

src/main/java/com/scalar/db/transaction/consensuscommit/Snapshot.java

@@ -100,6 +100,10 @@ public void put(Snapshot.Key key, Delete delete) {
     deleteSet.put(key, delete);
   }
 
+  public boolean containsKey(Snapshot.Key key) {
+    return writeSet.containsKey(key) || readSet.containsKey(key);
+  }
+
   public Optional<TransactionResult> get(Snapshot.Key key) {
     if (writeSet.containsKey(key)) {
       throw new CrudRuntimeException("reading already written data is not allowed");

src/test/java/com/scalar/db/transaction/consensuscommit/CrudHandlerTest.java

@@ -104,6 +104,7 @@ public void get_KeyExistsInSnapshot_ShouldReturnFromSnapshot() throws CrudExcept
     Get get = prepareGet();
     Optional<TransactionResult> expected =
         Optional.of(prepareResult(true, TransactionState.COMMITTED));
+    when(snapshot.containsKey(new Snapshot.Key(get))).thenReturn(true);
     when(snapshot.get(new Snapshot.Key(get))).thenReturn(expected);
 
     // Act
@@ -120,7 +121,7 @@ public void get_KeyExistsInSnapshot_ShouldReturnFromSnapshot() throws CrudExcept
     // Arrange
     Get get = prepareGet();
     Optional<Result> expected = Optional.of(prepareResult(true, TransactionState.COMMITTED));
-    when(snapshot.get(new Snapshot.Key(get))).thenReturn(Optional.empty());
+    when(snapshot.containsKey(new Snapshot.Key(get))).thenReturn(false);
     doNothing().when(snapshot).put(any(Snapshot.Key.class), any(Optional.class));
     when(storage.get(get)).thenReturn(expected);
 
@@ -270,9 +271,8 @@ public void scan_CalledTwice_SecondTimeShouldReturnTheSameFromSnapshot()
     when(snapshot.get(scan))
         .thenReturn(Optional.empty())
         .thenReturn(Optional.of(Arrays.asList(key)));
-    when(snapshot.get(key))
-        .thenReturn(Optional.empty())
-        .thenReturn(Optional.of((TransactionResult) result));
+    when(snapshot.containsKey(key)).thenReturn(false).thenReturn(true);
+    when(snapshot.get(key)).thenReturn(Optional.of((TransactionResult) result));
 
     // Act
     List<Result> results1 = handler.scan(scan);
@@ -332,9 +332,8 @@ public void scan_GetCalledAfterScan_ShouldReturnFromSnapshot()
             scan.getPartitionKey(),
             result.getClusteringKey().get());
     when(snapshot.get(scan)).thenReturn(Optional.empty());
-    when(snapshot.get(key))
-        .thenReturn(Optional.empty())
-        .thenReturn(Optional.of((TransactionResult) result));
+    when(snapshot.containsKey(key)).thenReturn(false).thenReturn(true);
+    when(snapshot.get(key)).thenReturn(Optional.of((TransactionResult) result));
 
     // Act
     List<Result> results = handler.scan(scan);