Implement scanner API for Consensus Commit

Author: brfrn169

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

@@ -911,6 +911,18 @@ public enum CoreError implements ScalarDbError {
       Category.USER_ERROR, "0203", "Delimiter must not be null", "", ""),
   DATA_LOADER_CONFIG_FILE_PATH_BLANK(
       Category.USER_ERROR, "0204", "Config file path must not be blank", "", ""),
+  CONSENSUS_COMMIT_SCANNER_NOT_CLOSED(
+      Category.USER_ERROR,
+      "0205",
+      "Some scanners were not closed. All scanners must be closed before committing the transaction.",
+      "",
+      ""),
+  TWO_PHASE_CONSENSUS_COMMIT_SCANNER_NOT_CLOSED(
+      Category.USER_ERROR,
+      "0206",
+      "Some scanners were not closed. All scanners must be closed before preparing the transaction.",
+      "",
+      ""),
 
   //
   // Errors for the concurrency error category

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

@@ -24,8 +24,10 @@
 import com.scalar.db.exception.transaction.UnsatisfiedConditionException;
 import com.scalar.db.util.ScalarDbUtils;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.slf4j.Logger;
@@ -98,7 +100,41 @@ public List<Result> scan(Scan scan) throws CrudException {
 
   @Override
   public Scanner getScanner(Scan scan) throws CrudException {
-    throw new UnsupportedOperationException("Implement later");
+    scan = copyAndSetTargetToIfNot(scan);
+    Scanner scanner = crud.getScanner(scan);
+
+    return new Scanner() {
+      @Override
+      public Optional<Result> one() throws CrudException {
+        try {
+          return scanner.one();
+        } catch (UncommittedRecordException e) {
+          lazyRecovery(e);
+          throw e;
+        }
+      }
+
+      @Override
+      public List<Result> all() throws CrudException {
+        try {
+          return scanner.all();
+        } catch (UncommittedRecordException e) {
+          lazyRecovery(e);
+          throw e;
+        }
+      }
+
+      @Override
+      public void close() throws CrudException {
+        scanner.close();
+      }
+
+      @Nonnull
+      @Override
+      public Iterator<Result> iterator() {
+        return scanner.iterator();
+      }
+    };
   }
 
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
@@ -213,6 +249,10 @@ public void mutate(List<? extends Mutation> mutations) throws CrudException {
 
   @Override
   public void commit() throws CommitException, UnknownTransactionStatusException {
+    if (!crud.areAllScannersClosed()) {
+      throw new IllegalStateException(CoreError.CONSENSUS_COMMIT_SCANNER_NOT_CLOSED.buildMessage());
+    }
+
     // Execute implicit pre-read
     try {
       crud.readIfImplicitPreReadEnabled();
@@ -234,6 +274,12 @@ public void commit() throws CommitException, UnknownTransactionStatusException {
 
   @Override
   public void rollback() {
+    try {
+      crud.closeScanners();
+    } catch (CrudException e) {
+      logger.warn("Failed to close the scanner", e);
+    }
+
     if (groupCommitter != null) {
       groupCommitter.remove(crud.getSnapshot().getId());
     }

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

@@ -16,10 +16,12 @@
 import com.scalar.db.api.Put;
 import com.scalar.db.api.Result;
 import com.scalar.db.api.Scan;
+import com.scalar.db.api.TransactionCrudOperable;
 import com.scalar.db.api.TransactionState;
 import com.scalar.db.api.Update;
 import com.scalar.db.api.Upsert;
 import com.scalar.db.common.AbstractDistributedTransactionManager;
+import com.scalar.db.common.AbstractTransactionManagerCrudOperableScanner;
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.exception.transaction.CommitConflictException;
 import com.scalar.db.exception.transaction.CrudConflictException;
@@ -34,6 +36,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 import org.slf4j.Logger;
@@ -231,9 +234,86 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
 
   @Override
   public Scanner getScanner(Scan scan) throws CrudException {
-    throw new UnsupportedOperationException("Implement later");
+    DistributedTransaction transaction = begin();
+
+    TransactionCrudOperable.Scanner scanner;
+    try {
+      scanner = transaction.getScanner(copyAndSetTargetToIfNot(scan));
+    } catch (CrudException e) {
+      rollbackTransaction(transaction);
+      throw e;
+    }
+
+    return new AbstractTransactionManagerCrudOperableScanner() {
+
+      private final AtomicBoolean closed = new AtomicBoolean();
+
+      @Override
+      public Optional<Result> one() throws CrudException {
+        try {
+          return scanner.one();
+        } catch (CrudException e) {
+          closed.set(true);
+
+          try {
+            scanner.close();
+          } catch (CrudException ex) {
+            e.addSuppressed(ex);
+          }
+
+          rollbackTransaction(transaction);
+          throw e;
+        }
+      }
+
+      @Override
+      public List<Result> all() throws CrudException {
+        try {
+          return scanner.all();
+        } catch (CrudException e) {
+          closed.set(true);
+
+          try {
+            scanner.close();
+          } catch (CrudException ex) {
+            e.addSuppressed(ex);
+          }
+
+          rollbackTransaction(transaction);
+          throw e;
+        }
+      }
+
+      @Override
+      public void close() throws CrudException, UnknownTransactionStatusException {
+        if (closed.get()) {
+          return;
+        }
+        closed.set(true);
+
+        try {
+          scanner.close();
+        } catch (CrudException e) {
+          rollbackTransaction(transaction);
+          throw e;
+        }
+
+        try {
+          transaction.commit();
+        } catch (CommitConflictException e) {
+          rollbackTransaction(transaction);
+          throw new CrudConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
+        } catch (UnknownTransactionStatusException e) {
+          throw e;
+        } catch (TransactionException e) {
+          rollbackTransaction(transaction);
+          throw new CrudException(e.getMessage(), e, e.getTransactionId().orElse(null));
+        }
+      }
+    };
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
   public void put(Put put) throws CrudException, UnknownTransactionStatusException {
@@ -244,6 +324,7 @@ public void put(Put put) throws CrudException, UnknownTransactionStatusException
         });
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
   public void put(List<Put> puts) throws CrudException, UnknownTransactionStatusException {
@@ -290,6 +371,7 @@ public void delete(Delete delete) throws CrudException, UnknownTransactionStatus
         });
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
   public void delete(List<Delete> deletes) throws CrudException, UnknownTransactionStatusException {