JdbcMetaStoreManagerFactory determines schemaVersion once per realm (#2217)

otherwise every call to `MetaStoreManagerFactory.getOrCreateSession` was
running the query.

Author: XN137

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -80,7 +80,7 @@ public class JdbcBasePersistenceImpl implements BasePersistence, IntegrationPers
   private final PrincipalSecretsGenerator secretsGenerator;
   private final PolarisStorageIntegrationProvider storageIntegrationProvider;
   private final String realmId;
-  private final int version;
+  private final int schemaVersion;
 
   // The max number of components a location can have before the optimized sibling check is not used
   private static final int MAX_LOCATION_COMPONENTS = 40;
@@ -89,12 +89,13 @@ public JdbcBasePersistenceImpl(
       DatasourceOperations databaseOperations,
       PrincipalSecretsGenerator secretsGenerator,
       PolarisStorageIntegrationProvider storageIntegrationProvider,
-      String realmId) {
+      String realmId,
+      int schemaVersion) {
     this.datasourceOperations = databaseOperations;
     this.secretsGenerator = secretsGenerator;
     this.storageIntegrationProvider = storageIntegrationProvider;
     this.realmId = realmId;
-    this.version = loadVersion();
+    this.schemaVersion = schemaVersion;
   }
 
   @Override
@@ -651,7 +652,7 @@ public boolean hasChildren(
     }
   }
 
-  private int loadVersion() {
+  static int loadSchemaVersion(DatasourceOperations datasourceOperations) {
     PreparedQuery query = QueryGenerator.generateVersionQuery();
     try {
       List<SchemaVersion> schemaVersion =
@@ -671,7 +672,7 @@ private int loadVersion() {
   public <T extends PolarisEntity & LocationBasedEntity>
       Optional<Optional<String>> hasOverlappingSiblings(
           @Nonnull PolarisCallContext callContext, T entity) {
-    if (this.version < 2) {
+    if (this.schemaVersion < 2) {
       return Optional.empty();
     }
     if (entity.getBaseLocation().chars().filter(ch -> ch == '/').count()

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -96,14 +96,17 @@ private void initializeForRealm(
     // Materialize realmId so that background tasks that don't have an active
     // RealmContext (request-scoped bean) can still create a JdbcBasePersistenceImpl
     String realmId = realmContext.getRealmIdentifier();
+    // determine schemaVersion once per realm
+    final int schemaVersion = JdbcBasePersistenceImpl.loadSchemaVersion(datasourceOperations);
     sessionSupplierMap.put(
         realmId,
         () ->
             new JdbcBasePersistenceImpl(
                 datasourceOperations,
                 secretsGenerator(realmId, rootCredentialsSet),
                 storageIntegrationProvider,
-                realmId));
+                realmId,
+                schemaVersion));
 
     PolarisMetaStoreManager metaStoreManager = createNewMetaStoreManager();
     metaStoreManagerMap.put(realmId, metaStoreManager);

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/AtomicMetastoreManagerWithJdbcBasePersistenceImplTest.java

@@ -45,6 +45,7 @@ public static DataSource createH2DataSource() {
 
   @Override
   protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
+    int schemaVersion = 2;
     PolarisDiagnostics diagServices = new PolarisDefaultDiagServiceImpl();
     DatasourceOperations datasourceOperations;
     try {
@@ -53,7 +54,7 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
       ClassLoader classLoader = DatasourceOperations.class.getClassLoader();
       InputStream scriptStream =
           classLoader.getResourceAsStream(
-              String.format("%s/schema-v2.sql", DatabaseType.H2.getDisplayName()));
+              String.format("%s/schema-v%s.sql", DatabaseType.H2.getDisplayName(), schemaVersion));
       datasourceOperations.executeScript(scriptStream);
     } catch (SQLException e) {
       throw new RuntimeException(
@@ -68,7 +69,8 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
             datasourceOperations,
             RANDOM_SECRETS,
             Mockito.mock(),
-            realmContext.getRealmIdentifier());
+            realmContext.getRealmIdentifier(),
+            schemaVersion);
     return new PolarisTestMetaStoreManager(
         new AtomicOperationMetaStoreManager(),
         new PolarisCallContext(