attempt refactor around datasourceOperations

Author: eric-maynard

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

@@ -98,15 +98,14 @@ protected PolarisMetaStoreManager createNewMetaStoreManager() {
   }
 
   private void initializeForRealm(
+      DatasourceOperations datasourceOperations,
       RealmContext realmContext,
-      RootCredentialsSet rootCredentialsSet,
-      Optional<SchemaOptions> schemaOptions) {
-    DatasourceOperations databaseOperations = getDatasourceOperations(schemaOptions);
+      RootCredentialsSet rootCredentialsSet) {
     sessionSupplierMap.put(
         realmContext.getRealmIdentifier(),
         () ->
             new JdbcBasePersistenceImpl(
-                databaseOperations,
+                datasourceOperations,
                 secretsGenerator(realmContext, rootCredentialsSet),
                 storageIntegrationProvider,
                 realmContext.getRealmIdentifier()));
@@ -115,24 +114,13 @@ private void initializeForRealm(
     metaStoreManagerMap.put(realmContext.getRealmIdentifier(), metaStoreManager);
   }
 
-  private DatasourceOperations getDatasourceOperations(Optional<SchemaOptions> schemaOptions) {
+  public DatasourceOperations getDatasourceOperations() {
     DatasourceOperations databaseOperations;
     try {
       databaseOperations = new DatasourceOperations(dataSource.get(), relationalJdbcConfiguration);
     } catch (SQLException sqlException) {
       throw new RuntimeException(sqlException);
     }
-    // If this is set, we are bootstrapping
-    if (schemaOptions.isPresent()) {
-      try {
-        // Run the set-up script to create the tables.
-        databaseOperations.executeScript(
-            databaseOperations.getDatabaseType().getInitScriptResource(schemaOptions.get()));
-      } catch (SQLException e) {
-        throw new RuntimeException(
-            String.format("Error executing sql script: %s", e.getMessage()), e);
-      }
-    }
     return databaseOperations;
   }
 
@@ -159,10 +147,19 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
     for (String realm : bootstrapOptions.realms()) {
       RealmContext realmContext = () -> realm;
       if (!metaStoreManagerMap.containsKey(realm)) {
+        DatasourceOperations datasourceOperations = getDatasourceOperations();
+        try {
+          // Run the set-up script to create the tables.
+          datasourceOperations.executeScript(
+            datasourceOperations.getDatabaseType().getInitScriptResource(bootstrapOptions.schemaOptions()));
+        } catch (SQLException e) {
+          throw new RuntimeException(
+            String.format("Error executing sql script: %s", e.getMessage()), e);
+        }
         initializeForRealm(
+          datasourceOperations,
             realmContext,
-            bootstrapOptions.rootCredentialsSet(),
-            Optional.of(bootstrapOptions.schemaOptions()));
+            bootstrapOptions.rootCredentialsSet());
         PrincipalSecretsResult secretsResult =
             bootstrapServiceAndCreatePolarisPrincipalForRealm(
                 realmContext, metaStoreManagerMap.get(realm));
@@ -198,7 +195,8 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
   public synchronized PolarisMetaStoreManager getOrCreateMetaStoreManager(
       RealmContext realmContext) {
     if (!metaStoreManagerMap.containsKey(realmContext.getRealmIdentifier())) {
-      initializeForRealm(realmContext, null, Optional.empty());
+      DatasourceOperations datasourceOperations = getDatasourceOperations();
+      initializeForRealm(datasourceOperations, realmContext, null);
       checkPolarisServiceBootstrappedForRealm(
           realmContext, metaStoreManagerMap.get(realmContext.getRealmIdentifier()));
     }
@@ -209,7 +207,8 @@ public synchronized PolarisMetaStoreManager getOrCreateMetaStoreManager(
   public synchronized Supplier<BasePersistence> getOrCreateSessionSupplier(
       RealmContext realmContext) {
     if (!sessionSupplierMap.containsKey(realmContext.getRealmIdentifier())) {
-      initializeForRealm(realmContext, null, Optional.empty());
+      DatasourceOperations datasourceOperations = getDatasourceOperations();
+      initializeForRealm(datasourceOperations, realmContext, null);
       checkPolarisServiceBootstrappedForRealm(
           realmContext, metaStoreManagerMap.get(realmContext.getRealmIdentifier()));
     } else {