Introduce DataSourceResolver for multi-datasource support in JDBC per…

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

@@ -0,0 +1,46 @@
+/*
+ * 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 javax.sql.DataSource;
+import org.apache.polaris.core.context.RealmContext;
+
+/**
+ * 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 {
+
+  /** The type of store representing the workload pattern. */
+  enum StoreType {
+    METASTORE,
+    METRICS,
+    EVENTS
+  }
+
+  /**
+   * Resolves the DataSource for a given realm and store type.
+   *
+   * @param realmContext the realm context
+   * @param storeType the type of store (e.g., main, metrics, events)
+   * @return the resolved DataSource
+   */
+  DataSource resolve(RealmContext realmContext, StoreType storeType);
+}

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

@@ -72,10 +72,15 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   final Map<String, Supplier<BasePersistence>> sessionSupplierMap = new HashMap<>();
 
   @Inject Clock clock;
+
   @Inject PolarisDiagnostics diagnostics;
+
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
-  @Inject Instance<DataSource> dataSource;
+
+  @Inject Instance<DataSourceResolver> dataSourceResolver;
+
   @Inject RelationalJdbcConfiguration relationalJdbcConfiguration;
+
   @Inject RealmConfig realmConfig;
 
   protected JdbcMetaStoreManagerFactory() {}
@@ -121,10 +126,12 @@ private void initializeForRealm(
     metaStoreManagerMap.put(realmId, metaStoreManager);
   }
 
-  public DatasourceOperations getDatasourceOperations() {
+  public DatasourceOperations getDatasourceOperations(
+      RealmContext realmContext, DataSourceResolver.StoreType storeType) {
     DatasourceOperations databaseOperations;
     try {
-      databaseOperations = new DatasourceOperations(dataSource.get(), relationalJdbcConfiguration);
+      DataSource resolvedDs = dataSourceResolver.get().resolve(realmContext, storeType);
+      databaseOperations = new DatasourceOperations(resolvedDs, relationalJdbcConfiguration);
     } catch (SQLException sqlException) {
       throw new RuntimeException(sqlException);
     }
@@ -154,7 +161,8 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
     for (String realm : bootstrapOptions.realms()) {
       RealmContext realmContext = () -> realm;
       if (!metaStoreManagerMap.containsKey(realm)) {
-        DatasourceOperations datasourceOperations = getDatasourceOperations();
+        DatasourceOperations datasourceOperations =
+            getDatasourceOperations(realmContext, DataSourceResolver.StoreType.METASTORE);
         int currentSchemaVersion =
             JdbcBasePersistenceImpl.loadSchemaVersion(datasourceOperations, true);
         int requestedSchemaVersion = JdbcBootstrapUtils.getRequestedSchemaVersion(bootstrapOptions);
@@ -219,7 +227,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, DataSourceResolver.StoreType.METASTORE);
       initializeForRealm(datasourceOperations, realmContext, null);
       checkPolarisServiceBootstrappedForRealm(realmContext);
     }
@@ -229,7 +238,8 @@ public synchronized PolarisMetaStoreManager getOrCreateMetaStoreManager(
   @Override
   public synchronized BasePersistence getOrCreateSession(RealmContext realmContext) {
     if (!sessionSupplierMap.containsKey(realmContext.getRealmIdentifier())) {
-      DatasourceOperations datasourceOperations = getDatasourceOperations();
+      DatasourceOperations datasourceOperations =
+          getDatasourceOperations(realmContext, DataSourceResolver.StoreType.METASTORE);
       initializeForRealm(datasourceOperations, realmContext, null);
     }
     checkPolarisServiceBootstrappedForRealm(realmContext);

runtime/common/build.gradle.kts

@@ -24,6 +24,7 @@ plugins {
 
 dependencies {
   compileOnly(libs.smallrye.config.core)
+  implementation(project(":polaris-core"))
   implementation(project(":polaris-relational-jdbc"))
   implementation(platform(libs.quarkus.amazon.services.bom))
   implementation("io.quarkiverse.amazonservices:quarkus-amazon-rds")

runtime/common/src/main/java/org/apache/polaris/quarkus/common/config/jdbc/DefaultDataSourceResolver.java

@@ -0,0 +1,55 @@
+/*
+ * 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.quarkus.common.config.jdbc;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import javax.sql.DataSource;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.persistence.relational.jdbc.DataSourceResolver;
+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.
+ */
+@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(RealmContext realmContext, StoreType storeType) {
+    LOGGER.debug(
+        "Using default DataSource for realm '{}' and store '{}'",
+        realmContext.getRealmIdentifier(),
+        storeType);
+    return defaultDataSource.get();
+  }
+}