Introduce DataSourceResolver for multi-datasource support in JDBC persistence

This commit introduces the DataSourceResolver interface and a
DefaultDataSourceResolver implementation to enable flexible routing
of DataSources based on realm and store type (e.g., main, metrics,
events).

Key changes:
- New DataSourceResolver interface with resolve(realmId, storeType)
  and getAllUniqueDataSources() methods
- DefaultDataSourceResolver that routes all requests to the default
  DataSource (backward-compatible no-op)
- Refactored JdbcMetaStoreManagerFactory to use DataSourceResolver
  instead of direct DataSource injection

This lays the groundwork for isolating different persistence workloads
(entity metadata vs metrics vs events) into separate connection pools
or databases to mitigate noisy neighbor effects.

Closes #3890

Author: Subham-KRLX

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

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.persistence.relational.jdbc;
+
+import java.util.Set;
+import javax.sql.DataSource;
+
+/**
+ * Service to resolve the correct {@link DataSource} for a given realm and store
+ * type.
+ * This enables isolating different workloads (e.g., entity metadata vs metrics
+ * vs events)
+ * into different physical databases or connection pools.
+ */
+public interface DataSourceResolver {
+
+    String STORE_TYPE_MAIN = "main";
+    String STORE_TYPE_METRICS = "metrics";
+    String STORE_TYPE_EVENTS = "events";
+
+    /**
+     * Resolves the DataSource for a given realm and store type.
+     *
+     * @param realmId   the realm identifier
+     * @param storeType the type of store (e.g., main, metrics, events)
+     * @return the resolved DataSource
+     */
+    DataSource resolve(String realmId, String storeType);
+
+    /**
+     * Returns all unique DataSources managed by this resolver.
+     * This is useful for global operations like schema initialization against all
+     * data sources.
+     *
+     * @return a set of all DataSources
+     */
+    Set<DataSource> getAllUniqueDataSources();
+}

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

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.persistence.relational.jdbc;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import java.util.HashSet;
+import java.util.Set;
+import javax.sql.DataSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Default implementation of {@link DataSourceResolver} that routes all realms
+ * and store types to a
+ * single default {@link DataSource}. This serves as both the production default
+ * and the base for
+ * multi-datasource extensions.
+ *
+ * <p>
+ * To enable per-realm or per-store datasource routing, this class can be
+ * extended or replaced
+ * with a custom implementation that resolves named datasources based on
+ * configuration.
+ */
+@ApplicationScoped
+public class DefaultDataSourceResolver implements DataSourceResolver {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultDataSourceResolver.class);
+
+    private final Instance<DataSource> defaultDataSource;
+
+    @Inject
+    public DefaultDataSourceResolver(Instance<DataSource> defaultDataSource) {
+        this.defaultDataSource = defaultDataSource;
+    }
+
+    @Override
+    public DataSource resolve(String realmId, String storeType) {
+        LOGGER.debug("Using default DataSource for realm '{}' and store '{}'", realmId, storeType);
+        return defaultDataSource.get();
+    }
+
+    @Override
+    public Set<DataSource> getAllUniqueDataSources() {
+        Set<DataSource> dataSources = new HashSet<>();
+        dataSources.add(defaultDataSource.get());
+        return dataSources;
+    }
+}

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

@@ -55,7 +55,8 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * The implementation of Configuration interface for configuring the {@link PolarisMetaStoreManager}
+ * The implementation of Configuration interface for configuring the
+ * {@link PolarisMetaStoreManager}
  * using a JDBC backed by SQL metastore. TODO: refactor - <a
  * href="https://github.com/apache/polaris/pull/1287/files#r2047487588">...</a>
  */
@@ -70,11 +71,15 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   final Map<String, Supplier<BasePersistence>> sessionSupplierMap = new HashMap<>();
   protected final PolarisDiagnostics diagServices = new PolarisDefaultDiagServiceImpl();
 
-  @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
-  @Inject Instance<DataSource> dataSource;
-  @Inject RelationalJdbcConfiguration relationalJdbcConfiguration;
+  @Inject
+  PolarisStorageIntegrationProvider storageIntegrationProvider;
+  @Inject
+  Instance<DataSourceResolver> dataSourceResolver;
+  @Inject
+  RelationalJdbcConfiguration relationalJdbcConfiguration;
 
-  protected JdbcMetaStoreManagerFactory() {}
+  protected JdbcMetaStoreManagerFactory() {
+  }
 
   protected PrincipalSecretsGenerator secretsGenerator(
       String realmId, @Nullable RootCredentialsSet rootCredentialsSet) {
@@ -98,21 +103,24 @@ private void initializeForRealm(
     String realmId = realmContext.getRealmIdentifier();
     sessionSupplierMap.put(
         realmId,
-        () ->
-            new JdbcBasePersistenceImpl(
-                datasourceOperations,
-                secretsGenerator(realmId, rootCredentialsSet),
-                storageIntegrationProvider,
-                realmId));
+        () -> new JdbcBasePersistenceImpl(
+            datasourceOperations, // TODO: We need a way to pass the right data source for the actual operation
+                                  // (metrics vs main),
+            // but for now JdbcBasePersistenceImpl does everything. So we pass the main
+            // DatasourceOperations.
+            secretsGenerator(realmId, rootCredentialsSet),
+            storageIntegrationProvider,
+            realmId));
 
     PolarisMetaStoreManager metaStoreManager = createNewMetaStoreManager();
     metaStoreManagerMap.put(realmId, metaStoreManager);
   }
 
-  public DatasourceOperations getDatasourceOperations() {
+  public DatasourceOperations getDatasourceOperations(String realmId, String storeType) {
     DatasourceOperations databaseOperations;
     try {
-      databaseOperations = new DatasourceOperations(dataSource.get(), relationalJdbcConfiguration);
+      DataSource resolvedDs = dataSourceResolver.get().resolve(realmId, storeType);
+      databaseOperations = new DatasourceOperations(resolvedDs, relationalJdbcConfiguration);
     } catch (SQLException sqlException) {
       throw new RuntimeException(sqlException);
     }
@@ -124,12 +132,11 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
       Iterable<String> realms, RootCredentialsSet rootCredentialsSet) {
     SchemaOptions schemaOptions = ImmutableSchemaOptions.builder().build();
 
-    BootstrapOptions bootstrapOptions =
-        ImmutableBootstrapOptions.builder()
-            .realms(realms)
-            .rootCredentialsSet(rootCredentialsSet)
-            .schemaOptions(schemaOptions)
-            .build();
+    BootstrapOptions bootstrapOptions = ImmutableBootstrapOptions.builder()
+        .realms(realms)
+        .rootCredentialsSet(rootCredentialsSet)
+        .schemaOptions(schemaOptions)
+        .build();
 
     return bootstrapRealms(bootstrapOptions);
   }
@@ -142,7 +149,7 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
     for (String realm : bootstrapOptions.realms()) {
       RealmContext realmContext = () -> realm;
       if (!metaStoreManagerMap.containsKey(realm)) {
-        DatasourceOperations datasourceOperations = getDatasourceOperations();
+        DatasourceOperations datasourceOperations = getDatasourceOperations(realm, DataSourceResolver.STORE_TYPE_MAIN);
         try {
           // Run the set-up script to create the tables.
           datasourceOperations.executeScript(
@@ -155,8 +162,7 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
         }
         initializeForRealm(
             datasourceOperations, realmContext, bootstrapOptions.rootCredentialsSet());
-        PrincipalSecretsResult secretsResult =
-            bootstrapServiceAndCreatePolarisPrincipalForRealm(realmContext);
+        PrincipalSecretsResult secretsResult = bootstrapServiceAndCreatePolarisPrincipalForRealm(realmContext);
         results.put(realm, secretsResult);
       }
     }
@@ -188,7 +194,8 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
   public synchronized PolarisMetaStoreManager getOrCreateMetaStoreManager(
       RealmContext realmContext) {
     if (!metaStoreManagerMap.containsKey(realmContext.getRealmIdentifier())) {
-      DatasourceOperations datasourceOperations = getDatasourceOperations();
+      DatasourceOperations datasourceOperations = getDatasourceOperations(realmContext.getRealmIdentifier(),
+          DataSourceResolver.STORE_TYPE_MAIN);
       initializeForRealm(datasourceOperations, realmContext, null);
       checkPolarisServiceBootstrappedForRealm(realmContext);
     }
@@ -198,7 +205,8 @@ public synchronized PolarisMetaStoreManager getOrCreateMetaStoreManager(
   @Override
   public synchronized BasePersistence getOrCreateSession(RealmContext realmContext) {
     if (!sessionSupplierMap.containsKey(realmContext.getRealmIdentifier())) {
-      DatasourceOperations datasourceOperations = getDatasourceOperations();
+      DatasourceOperations datasourceOperations = getDatasourceOperations(realmContext.getRealmIdentifier(),
+          DataSourceResolver.STORE_TYPE_MAIN);
       initializeForRealm(datasourceOperations, realmContext, null);
     }
     checkPolarisServiceBootstrappedForRealm(realmContext);
@@ -219,33 +227,30 @@ public synchronized EntityCache getOrCreateEntityCache(
   }
 
   /**
-   * This method bootstraps service for a given realm: i.e. creates all the needed entities in the
+   * This method bootstraps service for a given realm: i.e. creates all the needed
+   * entities in the
    * metastore and creates a root service principal.
    */
   private PrincipalSecretsResult bootstrapServiceAndCreatePolarisPrincipalForRealm(
       RealmContext realmContext) {
-    // While bootstrapping we need to act as a fake privileged context since the real
+    // While bootstrapping we need to act as a fake privileged context since the
+    // real
     // CallContext may not have been resolved yet.
-    PolarisMetaStoreManager metaStoreManager =
-        metaStoreManagerMap.get(realmContext.getRealmIdentifier());
+    PolarisMetaStoreManager metaStoreManager = metaStoreManagerMap.get(realmContext.getRealmIdentifier());
     BasePersistence metaStore = sessionSupplierMap.get(realmContext.getRealmIdentifier()).get();
-    PolarisCallContext polarisContext =
-        new PolarisCallContext(realmContext, metaStore, diagServices);
+    PolarisCallContext polarisContext = new PolarisCallContext(realmContext, metaStore, diagServices);
 
-    Optional<PrincipalEntity> preliminaryRootPrincipal =
-        metaStoreManager.findRootPrincipal(polarisContext);
+    Optional<PrincipalEntity> preliminaryRootPrincipal = metaStoreManager.findRootPrincipal(polarisContext);
     if (preliminaryRootPrincipal.isPresent()) {
-      String overrideMessage =
-          "It appears this metastore manager has already been bootstrapped. "
-              + "To continue bootstrapping, please first purge the metastore with the `purge` command.";
+      String overrideMessage = "It appears this metastore manager has already been bootstrapped. "
+          + "To continue bootstrapping, please first purge the metastore with the `purge` command.";
       LOGGER.error("\n\n {} \n\n", overrideMessage);
       throw new IllegalArgumentException(overrideMessage);
     }
 
     metaStoreManager.bootstrapPolarisService(polarisContext);
 
-    PrincipalEntity rootPrincipal =
-        metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
+    PrincipalEntity rootPrincipal = metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
     return metaStoreManager.loadPrincipalSecrets(
         polarisContext,
         rootPrincipal
@@ -254,18 +259,20 @@ private PrincipalSecretsResult bootstrapServiceAndCreatePolarisPrincipalForRealm
   }
 
   /**
-   * In this method we check if Service was bootstrapped for a given realm, i.e. that all the
-   * entities were created (root principal, root principal role, etc) If service was not
-   * bootstrapped we are throwing IllegalStateException exception That will cause service to crash
-   * and force user to run Bootstrap command and initialize MetaStore and create all the required
+   * In this method we check if Service was bootstrapped for a given realm, i.e.
+   * that all the
+   * entities were created (root principal, root principal role, etc) If service
+   * was not
+   * bootstrapped we are throwing IllegalStateException exception That will cause
+   * service to crash
+   * and force user to run Bootstrap command and initialize MetaStore and create
+   * all the required
    * entities
    */
   private void checkPolarisServiceBootstrappedForRealm(RealmContext realmContext) {
-    PolarisMetaStoreManager metaStoreManager =
-        metaStoreManagerMap.get(realmContext.getRealmIdentifier());
+    PolarisMetaStoreManager metaStoreManager = metaStoreManagerMap.get(realmContext.getRealmIdentifier());
     BasePersistence metaStore = sessionSupplierMap.get(realmContext.getRealmIdentifier()).get();
-    PolarisCallContext polarisContext =
-        new PolarisCallContext(realmContext, metaStore, diagServices);
+    PolarisCallContext polarisContext = new PolarisCallContext(realmContext, metaStore, diagServices);
 
     Optional<PrincipalEntity> rootPrincipal = metaStoreManager.findRootPrincipal(polarisContext);
     if (rootPrincipal.isEmpty()) {