Backport to branch(3) : Add isConsistentVirtualTableReadGuaranteed() to StorageInfo (#3208)

Co-authored-by: Toshihiro Suzuki <[email protected]>

Author: feeblefakie

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 isConsistentVirtualTableReadGuaranteed();
+
   /**
    * 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 consistentVirtualTableReadGuaranteed;
 
   public StorageInfoImpl(
       String storageName,
       MutationAtomicityUnit mutationAtomicityUnit,
-      int maxAtomicMutationsCount) {
+      int maxAtomicMutationsCount,
+      boolean consistentVirtualTableReadGuaranteed) {
     this.storageName = storageName;
     this.mutationAtomicityUnit = mutationAtomicityUnit;
     this.maxAtomicMutationsCount = maxAtomicMutationsCount;
+    this.consistentVirtualTableReadGuaranteed = consistentVirtualTableReadGuaranteed;
   }
 
   @Override
@@ -36,6 +39,11 @@ public int getMaxAtomicMutationsCount() {
     return maxAtomicMutationsCount;
   }
 
+  @Override
+  public boolean isConsistentVirtualTableReadGuaranteed() {
+    return consistentVirtualTableReadGuaranteed;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -47,12 +55,18 @@ public boolean equals(Object o) {
     StorageInfoImpl that = (StorageInfoImpl) o;
     return getMaxAtomicMutationsCount() == that.getMaxAtomicMutationsCount()
         && Objects.equals(getStorageName(), that.getStorageName())
-        && getMutationAtomicityUnit() == that.getMutationAtomicityUnit();
+        && getMutationAtomicityUnit() == that.getMutationAtomicityUnit()
+        && isConsistentVirtualTableReadGuaranteed()
+            == that.isConsistentVirtualTableReadGuaranteed();
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(getStorageName(), getMutationAtomicityUnit(), getMaxAtomicMutationsCount());
+    return Objects.hash(
+        getStorageName(),
+        getMutationAtomicityUnit(),
+        getMaxAtomicMutationsCount(),
+        isConsistentVirtualTableReadGuaranteed());
   }
 
   @Override
@@ -61,6 +75,7 @@ public String toString() {
         .add("storageName", storageName)
         .add("mutationAtomicityUnit", mutationAtomicityUnit)
         .add("maxAtomicMutationsCount", maxAtomicMutationsCount)
+        .add("consistentVirtualTableReadGuaranteed", consistentVirtualTableReadGuaranteed)
         .toString();
   }
 }

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

@@ -49,7 +49,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 systemNamespace;

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

@@ -71,7 +71,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

@@ -162,7 +162,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;
@@ -984,8 +978,22 @@ public Set<String> getNamespaceNames() throws ExecutionException {
   }
 
   @Override
-  public StorageInfo getStorageInfo(String namespace) {
-    return STORAGE_INFO;
+  public StorageInfo getStorageInfo(String namespace) throws ExecutionException {
+    boolean consistentVirtualTableReadGuaranteed;
+    try (Connection connection = dataSource.getConnection()) {
+      int isolationLevel = connection.getTransactionIsolation();
+      consistentVirtualTableReadGuaranteed =
+          isolationLevel >= rdbEngine.getMinimumIsolationLevelForConsistentVirtualTableRead();
+    } 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,
+        consistentVirtualTableReadGuaranteed);
   }
 
   @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 getMinimumIsolationLevelForConsistentVirtualTableRead() {
+    // 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

@@ -513,4 +513,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 getMinimumIsolationLevelForConsistentVirtualTableRead() {
+    // 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 getMinimumIsolationLevelForConsistentVirtualTableRead() {
+    // In Oracle, only the SERIALIZABLE isolation level guarantees consistent reads
+    return Connection.TRANSACTION_SERIALIZABLE;
+  }
 }