Support begin in read-only mode for JDBC transactions

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

@@ -7,8 +7,6 @@
 import java.util.Properties;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.EnumSource;
 
 public class JdbcTransactionIntegrationTest extends DistributedTransactionIntegrationTestBase {
 
@@ -44,16 +42,4 @@ public void abort_forOngoingTransaction_ShouldAbortCorrectly() {}
   @Override
   @Test
   public void rollback_forOngoingTransaction_ShouldRollbackCorrectly() {}
-
-  @Disabled("Implement later")
-  @Override
-  @Test
-  public void get_GetGivenForCommittedRecord_InReadOnlyMode_ShouldReturnRecord() {}
-
-  @Disabled("Implement later")
-  @Override
-  @ParameterizedTest
-  @EnumSource(ScanType.class)
-  public void scanOrGetScanner_ScanGivenForCommittedRecord_InReadOnlyMode_ShouldReturnRecords(
-      ScanType scanType) {}
 }

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

@@ -438,7 +438,7 @@ public TableMetadata getTableMetadata(String namespace, String table) throws Exe
     boolean tableExists = false;
 
     try (Connection connection = dataSource.getConnection()) {
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
 
       try (PreparedStatement preparedStatement =
           connection.prepareStatement(getSelectColumnsStatement())) {
@@ -510,7 +510,7 @@ public TableMetadata getImportTableMetadata(
     }
 
     try (Connection connection = dataSource.getConnection()) {
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
 
       String catalogName = rdbEngine.getCatalogName(namespace);
       String schemaName = rdbEngine.getSchemaName(namespace);
@@ -608,7 +608,7 @@ public Set<String> getNamespaceTableNames(String namespace) throws ExecutionExce
             + enclose(METADATA_COL_FULL_TABLE_NAME)
             + " LIKE ?";
     try (Connection connection = dataSource.getConnection()) {
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
 
       try (PreparedStatement preparedStatement =
           connection.prepareStatement(selectTablesOfNamespaceStatement)) {
@@ -644,7 +644,7 @@ public boolean namespaceExists(String namespace) throws ExecutionException {
             + enclose(NAMESPACE_COL_NAMESPACE_NAME)
             + " = ?";
     try (Connection connection = dataSource.getConnection()) {
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
 
       try (PreparedStatement statement = connection.prepareStatement(selectQuery)) {
         statement.setString(1, namespace);
@@ -992,7 +992,7 @@ private String encloseFullTableName(String schema, String table) {
   @Override
   public Set<String> getNamespaceNames() throws ExecutionException {
     try (Connection connection = dataSource.getConnection()) {
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
 
       String selectQuery =
           "SELECT * FROM " + encloseFullTableName(metadataSchema, NAMESPACES_TABLE);

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

@@ -82,7 +82,7 @@ public Optional<Result> get(Get get) throws ExecutionException {
     Connection connection = null;
     try {
       connection = dataSource.getConnection();
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
       return jdbcService.get(get, connection);
     } catch (SQLException e) {
       throw new ExecutionException(
@@ -98,7 +98,7 @@ public Scanner scan(Scan scan) throws ExecutionException {
     Connection connection = null;
     try {
       connection = dataSource.getConnection();
-      rdbEngine.setReadOnly(connection, true);
+      rdbEngine.setConnectionToReadOnly(connection, true);
       return jdbcService.getScanner(scan, connection);
     } catch (SQLException e) {
       close(connection);

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

@@ -340,7 +340,7 @@ public RdbEngineTimeTypeStrategy<Integer, Long, Long, Long> getTimeTypeStrategy(
   }
 
   @Override
-  public void setReadOnly(Connection connection, boolean readOnly) {
+  public void setConnectionToReadOnly(Connection connection, boolean readOnly) {
     // Do nothing. SQLite does not support read-only mode.
   }
 }

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

@@ -230,7 +230,8 @@ default void throwIfDuplicatedIndexWarning(SQLWarning warning) throws SQLExcepti
     // Do nothing
   }
 
-  default void setReadOnly(Connection connection, boolean readOnly) throws SQLException {
+  default void setConnectionToReadOnly(Connection connection, boolean readOnly)
+      throws SQLException {
     connection.setReadOnly(readOnly);
   }
 }

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

@@ -18,6 +18,7 @@
 import com.scalar.db.api.Upsert;
 import com.scalar.db.common.AbstractDistributedTransactionManager;
 import com.scalar.db.common.AbstractTransactionManagerCrudOperableScanner;
+import com.scalar.db.common.ReadOnlyDistributedTransaction;
 import com.scalar.db.common.TableMetadataManager;
 import com.scalar.db.common.checker.OperationChecker;
 import com.scalar.db.common.error.CoreError;
@@ -36,6 +37,7 @@
 import com.scalar.db.storage.jdbc.RdbEngineFactory;
 import com.scalar.db.storage.jdbc.RdbEngineStrategy;
 import com.scalar.db.util.ThrowableFunction;
+import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Optional;
@@ -90,14 +92,39 @@ public JdbcTransactionManager(DatabaseConfig databaseConfig) {
   @Override
   public DistributedTransaction begin() throws TransactionException {
     String txId = UUID.randomUUID().toString();
-    return begin(txId);
+    return begin(txId, false);
   }
 
   @Override
   public DistributedTransaction begin(String txId) throws TransactionException {
+    return begin(txId, false);
+  }
+
+  @Override
+  public DistributedTransaction beginReadOnly() throws TransactionException {
+    String txId = UUID.randomUUID().toString();
+    return begin(txId, true);
+  }
+
+  @Override
+  public DistributedTransaction beginReadOnly(String txId) throws TransactionException {
+    return begin(txId, true);
+  }
+
+  private DistributedTransaction begin(String txId, boolean readOnly) throws TransactionException {
     try {
-      JdbcTransaction transaction =
-          new JdbcTransaction(txId, jdbcService, dataSource.getConnection(), rdbEngine);
+      Connection connection = dataSource.getConnection();
+
+      DistributedTransaction transaction;
+      if (readOnly) {
+        rdbEngine.setConnectionToReadOnly(connection, true);
+        transaction =
+            new ReadOnlyDistributedTransaction(
+                new JdbcTransaction(txId, jdbcService, connection, rdbEngine));
+      } else {
+        transaction = new JdbcTransaction(txId, jdbcService, connection, rdbEngine);
+      }
+
       getNamespace().ifPresent(transaction::withNamespace);
       getTable().ifPresent(transaction::withTable);
       return transaction;
@@ -109,16 +136,6 @@ public DistributedTransaction begin(String txId) throws TransactionException {
     }
   }
 
-  @Override
-  public DistributedTransaction beginReadOnly() {
-    throw new UnsupportedOperationException("implement later");
-  }
-
-  @Override
-  public DistributedTransaction beginReadOnly(String txId) {
-    throw new UnsupportedOperationException("implement later");
-  }
-
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
   @SuppressWarnings("InlineMeSuggester")
   @Deprecated
@@ -173,19 +190,19 @@ public DistributedTransaction start(
 
   @Override
   public Optional<Result> get(Get get) throws CrudException, UnknownTransactionStatusException {
-    return executeTransaction(t -> t.get(copyAndSetTargetToIfNot(get)));
+    return executeTransaction(t -> t.get(copyAndSetTargetToIfNot(get)), true);
   }
 
   @Override
   public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStatusException {
-    return executeTransaction(t -> t.scan(copyAndSetTargetToIfNot(scan)));
+    return executeTransaction(t -> t.scan(copyAndSetTargetToIfNot(scan)), true);
   }
 
   @Override
   public Scanner getScanner(Scan scan) throws CrudException {
     DistributedTransaction transaction;
     try {
-      transaction = begin();
+      transaction = beginReadOnly();
     } catch (TransactionNotFoundException e) {
       throw new CrudConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
     } catch (TransactionException e) {
@@ -277,7 +294,8 @@ public void put(Put put) throws CrudException, UnknownTransactionStatusException
         t -> {
           t.put(copyAndSetTargetToIfNot(put));
           return null;
-        });
+        },
+        false);
   }
 
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
@@ -288,7 +306,8 @@ public void put(List<Put> puts) throws CrudException, UnknownTransactionStatusEx
         t -> {
           t.put(copyAndSetTargetToIfNot(puts));
           return null;
-        });
+        },
+        false);
   }
 
   @Override
@@ -297,7 +316,8 @@ public void insert(Insert insert) throws CrudException, UnknownTransactionStatus
         t -> {
           t.insert(copyAndSetTargetToIfNot(insert));
           return null;
-        });
+        },
+        false);
   }
 
   @Override
@@ -306,7 +326,8 @@ public void upsert(Upsert upsert) throws CrudException, UnknownTransactionStatus
         t -> {
           t.upsert(copyAndSetTargetToIfNot(upsert));
           return null;
-        });
+        },
+        false);
   }
 
   @Override
@@ -315,7 +336,8 @@ public void update(Update update) throws CrudException, UnknownTransactionStatus
         t -> {
           t.update(copyAndSetTargetToIfNot(update));
           return null;
-        });
+        },
+        false);
   }
 
   @Override
@@ -324,7 +346,8 @@ public void delete(Delete delete) throws CrudException, UnknownTransactionStatus
         t -> {
           t.delete(copyAndSetTargetToIfNot(delete));
           return null;
-        });
+        },
+        false);
   }
 
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
@@ -335,7 +358,8 @@ public void delete(List<Delete> deletes) throws CrudException, UnknownTransactio
         t -> {
           t.delete(copyAndSetTargetToIfNot(deletes));
           return null;
-        });
+        },
+        false);
   }
 
   @Override
@@ -345,15 +369,21 @@ public void mutate(List<? extends Mutation> mutations)
         t -> {
           t.mutate(copyAndSetTargetToIfNot(mutations));
           return null;
-        });
+        },
+        false);
   }
 
   private <R> R executeTransaction(
-      ThrowableFunction<DistributedTransaction, R, TransactionException> throwableFunction)
+      ThrowableFunction<DistributedTransaction, R, TransactionException> throwableFunction,
+      boolean readOnly)
       throws CrudException, UnknownTransactionStatusException {
     DistributedTransaction transaction;
     try {
-      transaction = begin();
+      if (readOnly) {
+        transaction = beginReadOnly();
+      } else {
+        transaction = begin();
+      }
     } catch (TransactionNotFoundException e) {
       throw new CrudConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
     } catch (TransactionException e) {