Disallow scan operations after delete in Consensus Commit (#2610)

Author: brfrn169

core/src/main/java/com/scalar/db/common/error/CoreError.java

@@ -449,15 +449,17 @@ public enum CoreError implements ScalarDbError {
       "",
       ""),
   CONSENSUS_COMMIT_WRITING_ALREADY_DELETED_DATA_NOT_ALLOWED(
-      Category.USER_ERROR, "0104", "Writing already-deleted data is not allowed", "", ""),
-  CONSENSUS_COMMIT_GETTING_DATA_NEITHER_IN_READ_SET_NOR_DELETE_SET_NOT_ALLOWED(
       Category.USER_ERROR,
-      "0105",
-      "Getting data neither in the read set nor the delete set is not allowed",
+      "0104",
+      "Writing data already-deleted by the same transaction is not allowed",
+      "",
+      ""),
+  CONSENSUS_COMMIT_SCANNING_ALREADY_WRITTEN_OR_DELETED_DATA_NOT_ALLOWED(
+      Category.USER_ERROR,
+      "0106",
+      "Scanning data already-written or already-deleted by the same transaction is not allowed",
       "",
       ""),
-  CONSENSUS_COMMIT_READING_ALREADY_WRITTEN_DATA_NOT_ALLOWED(
-      Category.USER_ERROR, "0106", "Reading already-written data is not allowed", "", ""),
   CONSENSUS_COMMIT_TRANSACTION_NOT_VALIDATED_IN_SERIALIZABLE(
       Category.USER_ERROR,
       "0107",
@@ -667,9 +669,17 @@ public enum CoreError implements ScalarDbError {
       "",
       ""),
   CONSENSUS_COMMIT_INSERTING_ALREADY_WRITTEN_DATA_NOT_ALLOWED(
-      Category.USER_ERROR, "0146", "Inserting already-written data is not allowed", "", ""),
+      Category.USER_ERROR,
+      "0146",
+      "Inserting data already-written by the same transaction is not allowed",
+      "",
+      ""),
   CONSENSUS_COMMIT_DELETING_ALREADY_INSERTED_DATA_NOT_ALLOWED(
-      Category.USER_ERROR, "0147", "Deleting already-inserted data is not allowed", "", ""),
+      Category.USER_ERROR,
+      "0147",
+      "Deleting data already-inserted by the same transaction is not allowed",
+      "",
+      ""),
   DATA_LOADER_INVALID_COLUMN_NON_EXISTENT(
       Category.USER_ERROR,
       "0148",

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

@@ -79,7 +79,11 @@ public Optional<Result> get(Get originalGet) throws CrudException {
     Get get = (Get) prepareStorageSelection(originalGet);
     Snapshot.Key key = new Snapshot.Key(get);
     readUnread(key, get);
-    return createGetResult(key, get, originalProjections);
+
+    TableMetadata metadata = getTableMetadata(get);
+    return snapshot
+        .getResult(key, get)
+        .map(r -> new FilteredResult(r, originalProjections, metadata, isIncludeMetadataEnabled));
   }
 
   @VisibleForTesting
@@ -113,35 +117,22 @@ void read(Snapshot.Key key, Get get) throws CrudException {
         snapshot.getId());
   }
 
-  private Optional<Result> createGetResult(Snapshot.Key key, Get get, List<String> projections)
-      throws CrudException {
-    TableMetadata metadata = getTableMetadata(key.getNamespace(), key.getTable());
-    return snapshot
-        .getResult(key, get)
-        .map(r -> new FilteredResult(r, projections, metadata, isIncludeMetadataEnabled));
-  }
-
-  public List<Result> scan(Scan scan) throws CrudException {
-    List<Result> results = scanInternal(scan);
-
-    // We verify if this scan does not overlap previous writes using the actual scan result. Because
-    // we support arbitrary conditions in the where clause of a scan (not only ScanAll, but also
-    // Scan and ScanWithIndex), we cannot determine whether the scan results will include a record
-    // whose key is the same as the key specified in the previous writes, without knowing the
-    // obtained keys in the actual scan. With this check, users can avoid seeing unexpected scan
-    // results that have not included previous writes yet.
-    snapshot.verify(scan);
-
-    return results;
-  }
-
-  private List<Result> scanInternal(Scan originalScan) throws CrudException {
+  public List<Result> scan(Scan originalScan) throws CrudException {
     List<String> originalProjections = new ArrayList<>(originalScan.getProjections());
     Scan scan = (Scan) prepareStorageSelection(originalScan);
+    Map<Snapshot.Key, TransactionResult> results = scanInternal(scan);
+    snapshot.verifyNoOverlap(scan, results);
 
+    TableMetadata metadata = getTableMetadata(scan);
+    return results.values().stream()
+        .map(r -> new FilteredResult(r, originalProjections, metadata, isIncludeMetadataEnabled))
+        .collect(Collectors.toList());
+  }
+
+  private Map<Snapshot.Key, TransactionResult> scanInternal(Scan scan) throws CrudException {
     Optional<Map<Snapshot.Key, TransactionResult>> resultsInSnapshot = snapshot.getResults(scan);
     if (resultsInSnapshot.isPresent()) {
-      return createScanResults(scan, originalProjections, resultsInSnapshot.get());
+      return resultsInSnapshot.get();
     }
 
     Map<Snapshot.Key, TransactionResult> results = new LinkedHashMap<>();
@@ -151,23 +142,21 @@ private List<Result> scanInternal(Scan originalScan) throws CrudException {
       scanner = scanFromStorage(scan);
       for (Result r : scanner) {
         TransactionResult result = new TransactionResult(r);
-        if (!result.isCommitted()) {
-          throw new UncommittedRecordException(
-              scan,
-              result,
-              CoreError.CONSENSUS_COMMIT_READ_UNCOMMITTED_RECORD.buildMessage(),
-              snapshot.getId());
-        }
-
         Snapshot.Key key = new Snapshot.Key(scan, r);
-
-        // We always update the read set to create before image by using the latest record (result)
-        // because another conflicting transaction might have updated the record after this
-        // transaction read it first.
-        snapshot.putIntoReadSet(key, Optional.of(result));
-
-        snapshot.getResult(key).ifPresent(value -> results.put(key, value));
+        processScanResult(key, scan, result);
+        results.put(key, result);
+      }
+    } catch (RuntimeException e) {
+      Exception exception;
+      if (e.getCause() instanceof ExecutionException) {
+        exception = (ExecutionException) e.getCause();
+      } else {
+        exception = e;
       }
+      throw new CrudException(
+          CoreError.CONSENSUS_COMMIT_SCANNING_RECORDS_FROM_STORAGE_FAILED.buildMessage(),
+          exception,
+          snapshot.getId());
     } finally {
       if (scanner != null) {
         try {
@@ -177,19 +166,26 @@ private List<Result> scanInternal(Scan originalScan) throws CrudException {
         }
       }
     }
+
     snapshot.putIntoScanSet(scan, results);
 
-    return createScanResults(scan, originalProjections, results);
+    return results;
   }
 
-  private List<Result> createScanResults(
-      Scan scan, List<String> projections, Map<Snapshot.Key, TransactionResult> results)
+  private void processScanResult(Snapshot.Key key, Scan scan, TransactionResult result)
       throws CrudException {
-    assert scan.forNamespace().isPresent() && scan.forTable().isPresent();
-    TableMetadata metadata = getTableMetadata(scan.forNamespace().get(), scan.forTable().get());
-    return results.values().stream()
-        .map(r -> new FilteredResult(r, projections, metadata, isIncludeMetadataEnabled))
-        .collect(Collectors.toList());
+    if (!result.isCommitted()) {
+      throw new UncommittedRecordException(
+          scan,
+          result,
+          CoreError.CONSENSUS_COMMIT_READ_UNCOMMITTED_RECORD.buildMessage(),
+          snapshot.getId());
+    }
+
+    // We always update the read set to create before image by using the latest record (result)
+    // because another conflicting transaction might have updated the record after this
+    // transaction read it first.
+    snapshot.putIntoReadSet(key, Optional.of(result));
   }
 
   public void put(Put put) throws CrudException {
@@ -322,14 +318,14 @@ private TransactionTableMetadata getTransactionTableMetadata(Operation operation
     }
   }
 
-  private TableMetadata getTableMetadata(String namespace, String table) throws CrudException {
+  private TableMetadata getTableMetadata(Operation operation) throws CrudException {
     try {
       TransactionTableMetadata metadata =
-          tableMetadataManager.getTransactionTableMetadata(namespace, table);
+          tableMetadataManager.getTransactionTableMetadata(operation);
       if (metadata == null) {
+        assert operation.forFullTableName().isPresent();
         throw new IllegalArgumentException(
-            CoreError.TABLE_NOT_FOUND.buildMessage(
-                ScalarDbUtils.getFullTableName(namespace, table)));
+            CoreError.TABLE_NOT_FOUND.buildMessage(operation.forFullTableName().get()));
       }
       return metadata.getTableMetadata();
     } catch (ExecutionException e) {