Implement scanner API for JDBC transactions

core/src/integration-test/java/com/scalar/db/transaction/jdbc/JdbcTransactionIntegrationTest.java

@@ -42,14 +42,4 @@ public void abort_forOngoingTransaction_ShouldAbortCorrectly() {}
   @Override
   @Test
   public void rollback_forOngoingTransaction_ShouldRollbackCorrectly() {}
-
-  @Disabled("Implement later")
-  @Override
-  @Test
-  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
-
-  @Disabled("Implement later")
-  @Override
-  @Test
-  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
 }

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

@@ -1182,6 +1182,8 @@ public enum CoreError implements ScalarDbError {
       Category.INTERNAL_ERROR, "0052", "Failed to read JSON file. Details: %s.", "", ""),
   DATA_LOADER_JSONLINES_FILE_READ_FAILED(
       Category.INTERNAL_ERROR, "0053", "Failed to read JSON Lines file. Details: %s.", "", ""),
+  JDBC_TRANSACTION_GETTING_SCANNER_FAILED(
+      Category.INTERNAL_ERROR, "0054", "Getting the scanner failed. Details: %s", "", ""),
 
   //
   // Errors for the unknown transaction status error category

core/src/main/java/com/scalar/db/storage/jdbc/JdbcService.java

@@ -96,9 +96,14 @@ public Optional<Result> get(Get get, Connection connection)
     }
   }
 
-  @SuppressFBWarnings("OBL_UNSATISFIED_OBLIGATION_EXCEPTION_EDGE")
   public Scanner getScanner(Scan scan, Connection connection)
       throws SQLException, ExecutionException {
+    return getScanner(scan, connection, true);
+  }
+
+  @SuppressFBWarnings("OBL_UNSATISFIED_OBLIGATION_EXCEPTION_EDGE")
+  public Scanner getScanner(Scan scan, Connection connection, boolean closeConnectionOnScannerClose)
+      throws SQLException, ExecutionException {
     operationChecker.check(scan);
 
     TableMetadata tableMetadata = tableMetadataManager.getTableMetadata(scan);
@@ -111,7 +116,8 @@ public Scanner getScanner(Scan scan, Connection connection)
         new ResultInterpreter(scan.getProjections(), tableMetadata, rdbEngine),
         connection,
         preparedStatement,
-        resultSet);
+        resultSet,
+        closeConnectionOnScannerClose);
   }
 
   public List<Result> scan(Scan scan, Connection connection)

core/src/main/java/com/scalar/db/storage/jdbc/ScannerImpl.java

@@ -25,17 +25,20 @@ public class ScannerImpl extends AbstractScanner {
   private final Connection connection;
   private final PreparedStatement preparedStatement;
   private final ResultSet resultSet;
+  private final boolean closeConnectionOnClose;
 
   @SuppressFBWarnings("EI_EXPOSE_REP2")
   public ScannerImpl(
       ResultInterpreter resultInterpreter,
       Connection connection,
       PreparedStatement preparedStatement,
-      ResultSet resultSet) {
+      ResultSet resultSet,
+      boolean closeConnectionOnClose) {
     this.resultInterpreter = Objects.requireNonNull(resultInterpreter);
     this.connection = Objects.requireNonNull(connection);
     this.preparedStatement = Objects.requireNonNull(preparedStatement);
     this.resultSet = Objects.requireNonNull(resultSet);
+    this.closeConnectionOnClose = closeConnectionOnClose;
   }
 
   @Override
@@ -75,10 +78,13 @@ public void close() {
     } catch (SQLException e) {
       logger.warn("Failed to close the preparedStatement", e);
     }
-    try {
-      connection.close();
-    } catch (SQLException e) {
-      logger.warn("Failed to close the connection", e);
+
+    if (closeConnectionOnClose) {
+      try {
+        connection.close();
+      } catch (SQLException e) {
+        logger.warn("Failed to close the connection", e);
+      }
     }
   }
 }

core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransaction.java

@@ -20,6 +20,7 @@
 import com.scalar.db.api.UpdateIfExists;
 import com.scalar.db.api.Upsert;
 import com.scalar.db.common.AbstractDistributedTransaction;
+import com.scalar.db.common.AbstractTransactionCrudOperableScanner;
 import com.scalar.db.common.error.CoreError;
 import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.exception.transaction.CommitConflictException;
@@ -32,6 +33,7 @@
 import com.scalar.db.storage.jdbc.JdbcService;
 import com.scalar.db.storage.jdbc.RdbEngineStrategy;
 import com.scalar.db.util.ScalarDbUtils;
+import java.io.IOException;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.List;
@@ -96,7 +98,46 @@ public List<Result> scan(Scan scan) throws CrudException {
 
   @Override
   public Scanner getScanner(Scan scan) throws CrudException {
-    throw new UnsupportedOperationException("Implement later");
+    scan = copyAndSetTargetToIfNot(scan);
+
+    com.scalar.db.api.Scanner scanner;
+    try {
+      scanner = jdbcService.getScanner(scan, connection, false);
+    } catch (SQLException e) {
+      throw createCrudException(
+          e, CoreError.JDBC_TRANSACTION_GETTING_SCANNER_FAILED.buildMessage(e.getMessage()));
+    } catch (ExecutionException e) {
+      throw new CrudException(e.getMessage(), e, txId);
+    }
+
+    return new AbstractTransactionCrudOperableScanner() {
+      @Override
+      public Optional<Result> one() throws CrudException {
+        try {
+          return scanner.one();
+        } catch (ExecutionException e) {
+          throw new CrudException(e.getMessage(), e, txId);
+        }
+      }
+
+      @Override
+      public List<Result> all() throws CrudException {
+        try {
+          return scanner.all();
+        } catch (ExecutionException e) {
+          throw new CrudException(e.getMessage(), e, txId);
+        }
+      }
+
+      @Override
+      public void close() throws CrudException {
+        try {
+          scanner.close();
+        } catch (IOException e) {
+          throw new CrudException(e.getMessage(), e, txId);
+        }
+      }
+    };
   }
 
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */

core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionManager.java

@@ -12,10 +12,12 @@
 import com.scalar.db.api.Result;
 import com.scalar.db.api.Scan;
 import com.scalar.db.api.SerializableStrategy;
+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.common.TableMetadataManager;
 import com.scalar.db.common.checker.OperationChecker;
 import com.scalar.db.common.error.CoreError;
@@ -38,6 +40,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.concurrent.ThreadSafe;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.slf4j.Logger;
@@ -170,9 +173,93 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
 
   @Override
   public Scanner getScanner(Scan scan) throws CrudException {
-    throw new UnsupportedOperationException("Implement later");
+    DistributedTransaction transaction;
+    try {
+      transaction = begin();
+    } catch (TransactionNotFoundException e) {
+      throw new CrudConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
+    } catch (TransactionException e) {
+      throw new CrudException(e.getMessage(), e, e.getTransactionId().orElse(null));
+    }
+
+    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 {
@@ -183,6 +270,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 {
@@ -229,6 +317,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 {

core/src/test/java/com/scalar/db/storage/jdbc/JdbcDatabaseTest.java

@@ -98,7 +98,8 @@ public void whenGetOperationExecuted_shouldCallJdbcService() throws Exception {
   public void whenScanOperationExecutedAndScannerClosed_shouldCallJdbcService() throws Exception {
     // Arrange
     when(jdbcService.getScanner(any(), any()))
-        .thenReturn(new ScannerImpl(resultInterpreter, connection, preparedStatement, resultSet));
+        .thenReturn(
+            new ScannerImpl(resultInterpreter, connection, preparedStatement, resultSet, true));
 
     // Act
     Scan scan = new Scan(new Key("p1", "val")).forNamespace(NAMESPACE).forTable(TABLE);