Optimize JDBC adaptor for MySQL

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

@@ -107,6 +107,8 @@ public class JdbcConfig {
   // is 1970-01-01.
   public static final String DEFAULT_DB2_TIME_COLUMN_DEFAULT_DATE_COMPONENT = "1970-01-01";
 
+  private final DatabaseConfig databaseConfig;
+
   private final String jdbcUrl;
   @Nullable private final String username;
   @Nullable private final String password;
@@ -136,6 +138,7 @@ public class JdbcConfig {
   private final LocalDate db2TimeColumnDefaultDateComponent;
 
   public JdbcConfig(DatabaseConfig databaseConfig) {
+    this.databaseConfig = databaseConfig;
     String storage = databaseConfig.getStorage();
     String transactionManager = databaseConfig.getTransactionManager();
     if (!storage.equals(STORAGE_NAME) && !transactionManager.equals(TRANSACTION_MANAGER_NAME)) {
@@ -278,6 +281,10 @@ public JdbcConfig(DatabaseConfig databaseConfig) {
     }
   }
 
+  public DatabaseConfig getDatabaseConfig() {
+    return databaseConfig;
+  }
+
   public String getJdbcUrl() {
     return jdbcUrl;
   }

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

@@ -70,7 +70,7 @@ public static BasicDataSource initDataSource(
     dataSource.setMaxTotal(config.getConnectionPoolMaxTotal());
     dataSource.setPoolPreparedStatements(config.isPreparedStatementsPoolEnabled());
     dataSource.setMaxOpenPreparedStatements(config.getPreparedStatementsPoolMaxOpen());
-    for (Entry<String, String> entry : rdbEngine.getConnectionProperties().entrySet()) {
+    for (Entry<String, String> entry : rdbEngine.getConnectionProperties(config).entrySet()) {
       dataSource.addConnectionProperty(entry.getKey(), entry.getValue());
     }
 
@@ -97,7 +97,7 @@ public static BasicDataSource initDataSourceForTableMetadata(
     dataSource.setMinIdle(config.getTableMetadataConnectionPoolMinIdle());
     dataSource.setMaxIdle(config.getTableMetadataConnectionPoolMaxIdle());
     dataSource.setMaxTotal(config.getTableMetadataConnectionPoolMaxTotal());
-    for (Entry<String, String> entry : rdbEngine.getConnectionProperties().entrySet()) {
+    for (Entry<String, String> entry : rdbEngine.getConnectionProperties(config).entrySet()) {
       dataSource.addConnectionProperty(entry.getKey(), entry.getValue());
     }
 
@@ -124,7 +124,7 @@ public static BasicDataSource initDataSourceForAdmin(
     dataSource.setMinIdle(config.getAdminConnectionPoolMinIdle());
     dataSource.setMaxIdle(config.getAdminConnectionPoolMaxIdle());
     dataSource.setMaxTotal(config.getAdminConnectionPoolMaxTotal());
-    for (Entry<String, String> entry : rdbEngine.getConnectionProperties().entrySet()) {
+    for (Entry<String, String> entry : rdbEngine.getConnectionProperties(config).entrySet()) {
       dataSource.addConnectionProperty(entry.getKey(), entry.getValue());
     }
     return dataSource;

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

@@ -432,7 +432,7 @@ public RdbEngineTimeTypeStrategy<String, LocalDateTime, String, String> getTimeT
   }
 
   @Override
-  public Map<String, String> getConnectionProperties() {
+  public Map<String, String> getConnectionProperties(JdbcConfig config) {
     ImmutableMap.Builder<String, String> props = new ImmutableMap.Builder<>();
     // With this Db2 will return a textual description of the error when
     // calling JDBC `SQLException.getMessage` instead of a short message containing only error codes

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

@@ -1,8 +1,12 @@
 package com.scalar.db.storage.jdbc;
 
 import com.scalar.db.io.DataType;
+import java.sql.Connection;
 import java.sql.Driver;
 import java.sql.JDBCType;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.Map;
 import javax.annotation.Nullable;
 
 class RdbEngineMariaDB extends RdbEngineMysql {
@@ -28,4 +32,14 @@ DataType getDataTypeForScalarDbInternal(
           type, typeName, columnSize, digits, columnDescription, overrideDataType);
     }
   }
+
+  @Override
+  public Map<String, String> getConnectionProperties(JdbcConfig config) {
+    return Collections.emptyMap();
+  }
+
+  @Override
+  public void setConnectionToReadOnly(Connection connection, boolean readOnly) throws SQLException {
+    connection.setReadOnly(readOnly);
+  }
 }

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

@@ -11,6 +11,7 @@
 import com.scalar.db.storage.jdbc.query.SelectQuery;
 import com.scalar.db.storage.jdbc.query.SelectWithLimitQuery;
 import com.scalar.db.storage.jdbc.query.UpsertQuery;
+import java.sql.Connection;
 import java.sql.Driver;
 import java.sql.JDBCType;
 import java.sql.ResultSet;
@@ -448,7 +449,19 @@ public TimestampTZColumn parseTimestampTZColumn(ResultSet resultSet, String colu
   }
 
   @Override
-  public Map<String, String> getConnectionProperties() {
+  public Map<String, String> getConnectionProperties(JdbcConfig config) {
+    if (config.getDatabaseConfig().getScanFetchSize() == Integer.MIN_VALUE) {
+      // If the scan fetch size is set to Integer.MIN_VALUE, use the streaming mode.
+      return Collections.emptyMap();
+    }
+
+    // Otherwise, use the cursor fetch mode.
     return Collections.singletonMap("useCursorFetch", "true");
   }
+
+  @Override
+  public void setConnectionToReadOnly(Connection connection, boolean readOnly) throws SQLException {
+    // Observed performance degradation when using read-only connections in MySQL. So we do not
+    // set the read-only mode for MySQL connections.
+  }
 }

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

@@ -375,7 +375,7 @@ public String tryAddIfNotExistsToCreateIndexSql(String createIndexSql) {
   }
 
   @Override
-  public Map<String, String> getConnectionProperties() {
+  public Map<String, String> getConnectionProperties(JdbcConfig config) {
     // Needed to keep the microsecond precision when sending the value of ScalarDB TIME type.
     // It is being considered setting to it to false by default in a future driver release.
     return ImmutableMap.of("sendTimeAsDatetime", "false");

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

@@ -205,9 +205,10 @@ default TimestampTZColumn parseTimestampTZColumn(ResultSet resultSet, String col
   /**
    * Return the connection properties for the underlying database.
    *
+   * @param config the JDBC configuration
    * @return a map where key=property_name and value=property_value
    */
-  default Map<String, String> getConnectionProperties() {
+  default Map<String, String> getConnectionProperties(JdbcConfig config) {
     return Collections.emptyMap();
   }
 

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

@@ -274,7 +274,7 @@ private void getTableMetadata_forX_ShouldReturnTableMetadata(
             .addSecondaryIndex("c4")
             .build();
     assertThat(actualMetadata).isEqualTo(expectedMetadata);
-    if (rdbEngine == RdbEngine.SQLITE) {
+    if (rdbEngine == RdbEngine.MYSQL || rdbEngine == RdbEngine.SQLITE) {
       verify(connection, never()).setReadOnly(anyBoolean());
     } else {
       verify(connection).setReadOnly(true);
@@ -2197,7 +2197,7 @@ private void getNamespaceTables_forX_ShouldReturnTableNames(
     Set<String> actualTableNames = admin.getNamespaceTableNames(namespace);
 
     // Assert
-    if (rdbEngine == RdbEngine.SQLITE) {
+    if (rdbEngine == RdbEngine.MYSQL || rdbEngine == RdbEngine.SQLITE) {
       verify(connection, never()).setReadOnly(anyBoolean());
     } else {
       verify(connection).setReadOnly(true);
@@ -2270,7 +2270,7 @@ private void namespaceExists_forXWithExistingNamespace_ShouldReturnTrue(
     // Assert
     assertThat(admin.namespaceExists(namespace)).isTrue();
 
-    if (rdbEngine == RdbEngine.SQLITE) {
+    if (rdbEngine == RdbEngine.MYSQL || rdbEngine == RdbEngine.SQLITE) {
       verify(connection, never()).setReadOnly(anyBoolean());
     } else {
       verify(connection).setReadOnly(true);
@@ -2965,7 +2965,7 @@ private void getNamespaceNames_forX_ShouldReturnNamespaceNames(
     Set<String> actualNamespaceNames = admin.getNamespaceNames();
 
     // Assert
-    if (rdbEngine == RdbEngine.SQLITE) {
+    if (rdbEngine == RdbEngine.MYSQL || rdbEngine == RdbEngine.SQLITE) {
       verify(connection, never()).setReadOnly(anyBoolean());
     } else {
       verify(connection).setReadOnly(true);
@@ -3050,7 +3050,11 @@ public Boolean answer(InvocationOnMock invocation) {
         .execute(expectedCheckTableExistStatement);
     assertThat(actual.getPartitionKeyNames()).hasSameElementsAs(ImmutableSet.of("pk1", "pk2"));
     assertThat(actual.getColumnDataTypes()).containsExactlyEntriesOf(expectedColumns);
-    verify(connection).setReadOnly(true);
+    if (rdbEngine == RdbEngine.MYSQL) {
+      verify(connection, never()).setReadOnly(anyBoolean());
+    } else {
+      verify(connection).setReadOnly(true);
+    }
     verify(rdbEngineStrategy)
         .getDataTypeForScalarDb(
             any(JDBCType.class),

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

@@ -62,7 +62,7 @@ public void setUp() throws Exception {
             databaseConfig,
             dataSource,
             tableMetadataDataSource,
-            RdbEngine.createRdbEngineStrategy(RdbEngine.MYSQL),
+            RdbEngine.createRdbEngineStrategy(RdbEngine.POSTGRESQL),
             jdbcService);
   }
 
@@ -382,7 +382,7 @@ public void mutate_withConflictError_shouldThrowRetriableExecutionException()
       throws SQLException, ExecutionException {
     // Arrange
     when(jdbcService.mutate(any(), any())).thenThrow(sqlException);
-    when(sqlException.getErrorCode()).thenReturn(1213);
+    when(sqlException.getSQLState()).thenReturn("40001");
 
     // Act Assert
     assertThatThrownBy(

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

@@ -51,7 +51,7 @@ public void initDataSource_NonTransactional_ShouldReturnProperDataSource() throw
     JdbcConfig config = new JdbcConfig(new DatabaseConfig(properties));
     Driver driver = new com.mysql.cj.jdbc.Driver();
     when(rdbEngine.getDriver()).thenReturn(driver);
-    when(rdbEngine.getConnectionProperties()).thenReturn(Collections.emptyMap());
+    when(rdbEngine.getConnectionProperties(config)).thenReturn(Collections.emptyMap());
 
     // Act
     BasicDataSource dataSource = JdbcUtils.initDataSource(config, rdbEngine);
@@ -95,7 +95,7 @@ public void initDataSource_Transactional_ShouldReturnProperDataSource() throws S
     JdbcConfig config = new JdbcConfig(new DatabaseConfig(properties));
     Driver driver = new org.postgresql.Driver();
     when(rdbEngine.getDriver()).thenReturn(driver);
-    when(rdbEngine.getConnectionProperties()).thenReturn(Collections.emptyMap());
+    when(rdbEngine.getConnectionProperties(config)).thenReturn(Collections.emptyMap());
 
     // Act
     BasicDataSource dataSource = JdbcUtils.initDataSource(config, rdbEngine, true);
@@ -135,7 +135,7 @@ public void initDataSource_WithRdbEngineConnectionProperties_ShouldAddProperties
     JdbcConfig config = new JdbcConfig(new DatabaseConfig(properties));
     Driver driver = new com.microsoft.sqlserver.jdbc.SQLServerDriver();
     when(rdbEngine.getDriver()).thenReturn(driver);
-    when(rdbEngine.getConnectionProperties())
+    when(rdbEngine.getConnectionProperties(config))
         .thenReturn(ImmutableMap.of("prop1", "prop1Value", "prop2", "prop2Value"));
 
     try (MockedStatic<JdbcUtils> jdbcUtils =

core/src/test/java/com/scalar/db/transaction/jdbc/JdbcTransactionManagerTest.java

@@ -79,7 +79,7 @@ public void setUp() throws Exception {
             databaseConfig,
             dataSource,
             tableMetadataDataSource,
-            RdbEngine.createRdbEngineStrategy(RdbEngine.MYSQL),
+            RdbEngine.createRdbEngineStrategy(RdbEngine.POSTGRESQL),
             jdbcService);
   }
 
@@ -208,7 +208,7 @@ public void get_withConflictError_shouldThrowCrudConflictException()
       throws SQLException, ExecutionException {
     // Arrange
     when(jdbcService.get(any(), any())).thenThrow(sqlException);
-    when(sqlException.getErrorCode()).thenReturn(1213);
+    when(sqlException.getSQLState()).thenReturn("40001");
 
     // Act Assert
     assertThatThrownBy(
@@ -241,7 +241,7 @@ public void scan_withConflictError_shouldThrowCrudConflictException()
       throws SQLException, ExecutionException {
     // Arrange
     when(jdbcService.scan(any(), any())).thenThrow(sqlException);
-    when(sqlException.getErrorCode()).thenReturn(1213);
+    when(sqlException.getSQLState()).thenReturn("40001");
 
     // Act Assert
     assertThatThrownBy(
@@ -641,7 +641,7 @@ public void put_withConflictError_shouldThrowCrudConflictException()
       throws SQLException, ExecutionException {
     // Arrange
     when(jdbcService.put(any(), any())).thenThrow(sqlException);
-    when(sqlException.getErrorCode()).thenReturn(1213);
+    when(sqlException.getSQLState()).thenReturn("40001");
 
     // Act Assert
     assertThatThrownBy(
@@ -680,7 +680,7 @@ public void delete_withConflictError_shouldThrowCrudConflictException()
       throws SQLException, ExecutionException {
     // Arrange
     when(jdbcService.delete(any(), any())).thenThrow(sqlException);
-    when(sqlException.getErrorCode()).thenReturn(1213);
+    when(sqlException.getSQLState()).thenReturn("40001");
 
     // Act Assert
     assertThatThrownBy(
@@ -721,7 +721,7 @@ public void mutate_withConflictError_shouldThrowCrudConflictException()
       throws SQLException, ExecutionException {
     // Arrange
     when(jdbcService.put(any(), any())).thenThrow(sqlException);
-    when(sqlException.getErrorCode()).thenReturn(1213);
+    when(sqlException.getSQLState()).thenReturn("40001");
 
     // Act Assert
     assertThatThrownBy(