Add isConsistentVirtualTableRead() to StorageInfo

Author: brfrn169

core/src/main/java/com/scalar/db/api/DistributedStorageAdmin.java

@@ -47,6 +47,9 @@ public interface DistributedStorageAdmin extends Admin, AutoCloseable {
   /**
    * Returns the storage information.
    *
+   * <p>Note: This feature is primarily for internal use. Breaking changes can and will be
+   * introduced to it. Users should not depend on it.
+   *
    * @param namespace the namespace to get the storage information for
    * @return the storage information
    * @throws ExecutionException if the operation fails

core/src/main/java/com/scalar/db/api/StorageInfo.java

@@ -22,6 +22,13 @@ public interface StorageInfo {
    */
   int getMaxAtomicMutationsCount();
 
+  /**
+   * Returns whether the storage guarantees consistent reads for virtual tables.
+   *
+   * @return true if the storage guarantees consistent reads for virtual tables, false otherwise
+   */
+  boolean isConsistentVirtualTableRead();
+
   /**
    * The mutation atomicity unit of the storage.
    *

core/src/main/java/com/scalar/db/common/StorageInfoImpl.java

@@ -11,14 +11,17 @@ public class StorageInfoImpl implements StorageInfo {
   private final String storageName;
   private final MutationAtomicityUnit mutationAtomicityUnit;
   private final int maxAtomicMutationsCount;
+  private final boolean consistentVirtualTableRead;
 
   public StorageInfoImpl(
       String storageName,
       MutationAtomicityUnit mutationAtomicityUnit,
-      int maxAtomicMutationsCount) {
+      int maxAtomicMutationsCount,
+      boolean consistentVirtualTableRead) {
     this.storageName = storageName;
     this.mutationAtomicityUnit = mutationAtomicityUnit;
     this.maxAtomicMutationsCount = maxAtomicMutationsCount;
+    this.consistentVirtualTableRead = consistentVirtualTableRead;
   }
 
   @Override
@@ -36,6 +39,11 @@ public int getMaxAtomicMutationsCount() {
     return maxAtomicMutationsCount;
   }
 
+  @Override
+  public boolean isConsistentVirtualTableRead() {
+    return consistentVirtualTableRead;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

core/src/main/java/com/scalar/db/storage/cassandra/CassandraAdmin.java

@@ -57,7 +57,8 @@ public class CassandraAdmin implements DistributedStorageAdmin {
           "cassandra",
           StorageInfo.MutationAtomicityUnit.PARTITION,
           // No limit on the number of mutations
-          Integer.MAX_VALUE);
+          Integer.MAX_VALUE,
+          false);
 
   private final ClusterManager clusterManager;
   private final String metadataKeyspace;

core/src/main/java/com/scalar/db/storage/cosmos/CosmosAdmin.java

@@ -74,7 +74,8 @@ public class CosmosAdmin implements DistributedStorageAdmin {
           "cosmos",
           StorageInfo.MutationAtomicityUnit.PARTITION,
           // No limit on the number of mutations
-          Integer.MAX_VALUE);
+          Integer.MAX_VALUE,
+          false);
 
   private final CosmosClient client;
   private final String metadataDatabase;

core/src/main/java/com/scalar/db/storage/dynamo/DynamoAdmin.java

@@ -165,7 +165,8 @@ public class DynamoAdmin implements DistributedStorageAdmin {
           "dynamo",
           StorageInfo.MutationAtomicityUnit.STORAGE,
           // DynamoDB has a limit of 100 items per transactional batch write operation
-          100);
+          100,
+          false);
 
   private final DynamoDbClient client;
   private final ApplicationAutoScalingClient applicationAutoScalingClient;

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

@@ -53,12 +53,6 @@ public class JdbcAdmin implements DistributedStorageAdmin {
   @VisibleForTesting static final String JDBC_COL_DECIMAL_DIGITS = "DECIMAL_DIGITS";
 
   private static final String INDEX_NAME_PREFIX = "index";
-  private static final StorageInfo STORAGE_INFO =
-      new StorageInfoImpl(
-          "jdbc",
-          StorageInfo.MutationAtomicityUnit.STORAGE,
-          // No limit on the number of mutations
-          Integer.MAX_VALUE);
 
   private final RdbEngineStrategy rdbEngine;
   private final BasicDataSource dataSource;
@@ -1011,8 +1005,22 @@ public void upgrade(Map<String, String> options) throws ExecutionException {
   }
 
   @Override
-  public StorageInfo getStorageInfo(String namespace) {
-    return STORAGE_INFO;
+  public StorageInfo getStorageInfo(String namespace) throws ExecutionException {
+    boolean consistentVirtualTableRead;
+    try (Connection connection = dataSource.getConnection()) {
+      int isolationLevel = connection.getTransactionIsolation();
+      consistentVirtualTableRead =
+          isolationLevel >= rdbEngine.getMinimumIsolationLevelForConsistencyReads();
+    } catch (SQLException e) {
+      throw new ExecutionException("Getting the transaction isolation level failed", e);
+    }
+
+    return new StorageInfoImpl(
+        "jdbc",
+        StorageInfo.MutationAtomicityUnit.STORAGE,
+        // No limit on the number of mutations
+        Integer.MAX_VALUE,
+        consistentVirtualTableRead);
   }
 
   @Override

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

@@ -20,6 +20,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;
@@ -588,4 +589,10 @@ public void throwIfCrossPartitionScanOrderingOnBlobColumnNotSupported(
   public String getTableNamesInNamespaceSql() {
     return "SELECT TABNAME FROM SYSCAT.TABLES WHERE TABSCHEMA = ? AND TYPE = 'T'";
   }
+
+  @Override
+  public int getMinimumIsolationLevelForConsistencyReads() {
+    // In Db2, REPEATABLE READ and SERIALIZABLE isolation levels guarantee consistent reads
+    return Connection.TRANSACTION_REPEATABLE_READ;
+  }
 }

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

@@ -509,4 +509,10 @@ public void setConnectionToReadOnly(Connection connection, boolean readOnly) thr
   public String getTableNamesInNamespaceSql() {
     return "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ?";
   }
+
+  @Override
+  public int getMinimumIsolationLevelForConsistencyReads() {
+    // In MySQL, REPEATABLE READ and SERIALIZABLE isolation levels guarantee consistent reads
+    return Connection.TRANSACTION_REPEATABLE_READ;
+  }
 }

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

@@ -18,6 +18,7 @@
 import com.scalar.db.storage.jdbc.query.UpsertQuery;
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.sql.Connection;
 import java.sql.Driver;
 import java.sql.JDBCType;
 import java.sql.PreparedStatement;
@@ -540,4 +541,10 @@ public void bindBlobColumnToPreparedStatement(
   public String getTableNamesInNamespaceSql() {
     return "SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER = ?";
   }
+
+  @Override
+  public int getMinimumIsolationLevelForConsistencyReads() {
+    // In Oracle, only the SERIALIZABLE isolation level guarantees consistent reads
+    return Connection.TRANSACTION_SERIALIZABLE;
+  }
 }