Add isConsistentVirtualTableReadGuaranteed() to StorageInfo

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

@@ -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 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

@@ -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 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;
+  }
 }

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

@@ -10,6 +10,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.SQLException;
@@ -400,4 +401,10 @@ public String tryAddIfNotExistsToCreateIndexSql(String createIndexSql) {
   public String getTableNamesInNamespaceSql() {
     return "SELECT table_name FROM information_schema.tables WHERE table_schema = ?";
   }
+
+  @Override
+  public int getMinimumIsolationLevelForConsistentVirtualTableRead() {
+    // In PostgreSQL, REPEATABLE READ and SERIALIZABLE isolation levels guarantee consistent reads
+    return Connection.TRANSACTION_REPEATABLE_READ;
+  }
 }

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

@@ -10,6 +10,7 @@
 import com.scalar.db.storage.jdbc.query.SelectQuery;
 import com.scalar.db.storage.jdbc.query.SelectWithTop;
 import com.scalar.db.storage.jdbc.query.UpsertQuery;
+import java.sql.Connection;
 import java.sql.Driver;
 import java.sql.JDBCType;
 import java.sql.SQLException;
@@ -436,4 +437,10 @@ public Map<String, String> getConnectionProperties(JdbcConfig config) {
   public String getTableNamesInNamespaceSql() {
     return "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ?";
   }
+
+  @Override
+  public int getMinimumIsolationLevelForConsistentVirtualTableRead() {
+    // In SQL Server, REPEATABLE READ or higher isolation level guarantees consistent reads
+    return Connection.TRANSACTION_REPEATABLE_READ;
+  }
 }

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

@@ -389,4 +389,10 @@ public String getTableNamesInNamespaceSql() {
     // Do nothing. Namespace is just a table prefix in the SQLite implementation.
     return null;
   }
+
+  @Override
+  public int getMinimumIsolationLevelForConsistentVirtualTableRead() {
+    // In SQLite, READ COMMITTED and higher isolation levels guarantee consistent reads
+    return Connection.TRANSACTION_READ_COMMITTED;
+  }
 }

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

@@ -325,4 +325,17 @@ default void throwIfConjunctionsColumnNotSupported(
       Set<Conjunction> conjunctions, TableMetadata metadata) {}
 
   String getTableNamesInNamespaceSql();
+
+  /**
+   * Returns the minimum isolation level required to ensure consistent reads across virtual tables.
+   *
+   * <p>A virtual table read involves querying multiple underlying source tables. When using a lower
+   * isolation level, there is a risk of observing an inconsistent snapshot where data from
+   * different source tables reflects different points in time. This method returns the minimum
+   * isolation level that guarantees a consistent snapshot across all source tables involved in a
+   * virtual table read.
+   *
+   * @return the minimum isolation level required for consistent virtual table reads
+   */
+  int getMinimumIsolationLevelForConsistentVirtualTableRead();
 }

core/src/main/java/com/scalar/db/storage/multistorage/MultiStorageAdmin.java

@@ -305,7 +305,8 @@ public StorageInfo getStorageInfo(String namespace) throws ExecutionException {
       return new StorageInfoImpl(
           holder.storageName,
           storageInfo.getMutationAtomicityUnit(),
-          storageInfo.getMaxAtomicMutationsCount());
+          storageInfo.getMaxAtomicMutationsCount(),
+          storageInfo.isConsistentVirtualTableReadGuaranteed());
     } catch (RuntimeException e) {
       if (e.getCause() instanceof ExecutionException) {
         throw (ExecutionException) e.getCause();

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageAdmin.java

@@ -37,7 +37,8 @@ public class ObjectStorageAdmin implements DistributedStorageAdmin {
           "object_storage",
           StorageInfo.MutationAtomicityUnit.PARTITION,
           // No limit on the number of mutations
-          Integer.MAX_VALUE);
+          Integer.MAX_VALUE,
+          false);
 
   private final ObjectStorageWrapper wrapper;
   private final String metadataNamespace;

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapper.java

@@ -76,6 +76,10 @@ public interface ObjectStorageWrapper {
    */
   void deleteByPrefix(String prefix) throws ObjectStorageWrapperException;
 
-  /** Close the storage wrapper. */
+  /**
+   * Close the storage wrapper.
+   *
+   * @throws ObjectStorageWrapperException if an error occurs
+   */
   void close() throws ObjectStorageWrapperException;
 }

core/src/test/java/com/scalar/db/common/checker/OperationCheckerTest.java

@@ -53,7 +53,7 @@ public class OperationCheckerTest {
   private static final String COL3 = "v3";
   private static final StorageInfo STORAGE_INFO =
       new StorageInfoImpl(
-          "cassandra", StorageInfo.MutationAtomicityUnit.PARTITION, Integer.MAX_VALUE);
+          "cassandra", StorageInfo.MutationAtomicityUnit.PARTITION, Integer.MAX_VALUE, false);
 
   @Mock private DatabaseConfig databaseConfig;
   @Mock private TableMetadataManager metadataManager;
@@ -2059,9 +2059,11 @@ public void check_MutationsGiven_ForAtomicityUnit_ShouldBehaveCorrectly(
                 .addClusteringKey(CKEY1)
                 .build());
 
-    StorageInfo storageInfo1 = new StorageInfoImpl("s1", mutationAtomicityUnit, Integer.MAX_VALUE);
+    StorageInfo storageInfo1 =
+        new StorageInfoImpl("s1", mutationAtomicityUnit, Integer.MAX_VALUE, false);
     StorageInfo storageInfo2 =
-        new StorageInfoImpl("s2", StorageInfo.MutationAtomicityUnit.STORAGE, Integer.MAX_VALUE);
+        new StorageInfoImpl(
+            "s2", StorageInfo.MutationAtomicityUnit.STORAGE, Integer.MAX_VALUE, false);
     when(storageInfoProvider.getStorageInfo("ns")).thenReturn(storageInfo1);
     when(storageInfoProvider.getStorageInfo("ns2")).thenReturn(storageInfo1);
     when(storageInfoProvider.getStorageInfo("other_ns")).thenReturn(storageInfo2);

core/src/test/java/com/scalar/db/storage/cosmos/CosmosOperationCheckerTest.java

@@ -41,7 +41,8 @@ public class CosmosOperationCheckerTest {
   private static final String COL1 = "v1";
   private static final String COL2 = "v2";
   private static final StorageInfo STORAGE_INFO =
-      new StorageInfoImpl("cosmos", StorageInfo.MutationAtomicityUnit.PARTITION, Integer.MAX_VALUE);
+      new StorageInfoImpl(
+          "cosmos", StorageInfo.MutationAtomicityUnit.PARTITION, Integer.MAX_VALUE, false);
 
   private static final TableMetadata TABLE_METADATA1 =
       TableMetadata.newBuilder()

core/src/test/java/com/scalar/db/storage/dynamo/DynamoOperationCheckerTest.java

@@ -39,7 +39,7 @@ public class DynamoOperationCheckerTest {
   private static final String COL3 = "v3";
   private static final String COL4 = "v4";
   private static final StorageInfo STORAGE_INFO =
-      new StorageInfoImpl("dynamo", StorageInfo.MutationAtomicityUnit.STORAGE, 100);
+      new StorageInfoImpl("dynamo", StorageInfo.MutationAtomicityUnit.STORAGE, 100, false);
 
   @Mock private DatabaseConfig databaseConfig;
   @Mock private TableMetadataManager metadataManager;