Support Runtime JDBC Driver Loading and Dynamic Datasource Creation

gradle/libs.versions.toml

@@ -68,6 +68,7 @@ hadoop-client-api = { module = "org.apache.hadoop:hadoop-client-api", version.re
 hadoop-client-runtime = { module = "org.apache.hadoop:hadoop-client-runtime", version.ref = "hadoop" }
 hadoop-common = { module = "org.apache.hadoop:hadoop-common", version.ref = "hadoop" }
 hawkular-agent-prometheus-scraper = { module = "org.hawkular.agent:prometheus-scraper", version = "0.23.0.Final" }
+hikari-cp = { module = "com.zaxxer:HikariCP", version = "7.1.0" }
 hive-metastore = { module = "org.apache.hive:hive-metastore", version = "3.1.3" }
 iceberg-bom = { module = "org.apache.iceberg:iceberg-bom", version.ref = "iceberg" }
 immutables-builder = { module = "org.immutables:builder", version.ref = "immutables" }

persistence/relational-jdbc/build.gradle.kts

@@ -26,6 +26,7 @@ dependencies {
   implementation(project(":polaris-core"))
   implementation(libs.slf4j.api)
   implementation(libs.guava)
+  implementation(libs.hikari.cp)
 
   implementation(platform(libs.jackson3.bom))
   implementation("tools.jackson.core:jackson-core")

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

@@ -49,7 +49,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class DatasourceOperations {
+public class DatasourceOperations implements AutoCloseable {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DatasourceOperations.class);
 
@@ -69,14 +69,23 @@ public class DatasourceOperations {
 
   private final DataSource datasource;
   private final RelationalJdbcConfiguration relationalJdbcConfiguration;
+  private final boolean closeDataSourceOnClose;
   private final DatabaseType databaseType;
 
   private static final Random random = new Random();
 
   public DatasourceOperations(
       DataSource datasource, RelationalJdbcConfiguration relationalJdbcConfiguration) {
+    this(datasource, relationalJdbcConfiguration, false);
+  }
+
+  DatasourceOperations(
+      DataSource datasource,
+      RelationalJdbcConfiguration relationalJdbcConfiguration,
+      boolean closeDataSourceOnClose) {
     this.datasource = datasource;
     this.relationalJdbcConfiguration = relationalJdbcConfiguration;
+    this.closeDataSourceOnClose = closeDataSourceOnClose;
     try (Connection connection = this.datasource.getConnection()) {
       // Get explicitly configured database type, if any
       DatabaseType configuredType =
@@ -98,6 +107,21 @@ DatabaseType getDatabaseType() {
     return databaseType;
   }
 
+  boolean ownsDataSource() {
+    return closeDataSourceOnClose;
+  }
+
+  @Override
+  public void close() {
+    if (closeDataSourceOnClose && datasource instanceof AutoCloseable closeable) {
+      try {
+        closeable.close();
+      } catch (Exception e) {
+        throw new RuntimeException("Failed to close datasource", e);
+      }
+    }
+  }
+
   /**
    * Execute SQL script and close the associated input stream
    *

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

@@ -0,0 +1,217 @@
+/*
+ * 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 com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
+import java.util.stream.Stream;
+import javax.sql.DataSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class JdbcDataSourceFactory {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(JdbcDataSourceFactory.class);
+
+  private static final String DEFAULT_DRIVER_DIRECTORY = "jdbc-drivers";
+  private static final String POOL_NAME = "polaris-relational-jdbc";
+
+  private JdbcDataSourceFactory() {}
+
+  static Optional<DataSource> create(RelationalJdbcConfiguration configuration) {
+    Optional<String> jdbcUrl = nonBlank(configuration.jdbcUrl());
+    if (jdbcUrl.isEmpty()) {
+      return Optional.empty();
+    }
+
+    Optional<URLClassLoader> driverClassLoader = createDriverClassLoader(configuration);
+    if (driverClassLoader.isPresent()) {
+      return Optional.of(
+          createHikariDataSource(configuration, jdbcUrl.get(), driverClassLoader.get()));
+    }
+
+    return Optional.of(new HikariDataSource(createHikariConfig(configuration, jdbcUrl.get())));
+  }
+
+  private static HikariDataSource createHikariDataSource(
+      RelationalJdbcConfiguration configuration, String jdbcUrl, URLClassLoader driverClassLoader) {
+    ClassLoader previousContextClassLoader = Thread.currentThread().getContextClassLoader();
+    boolean success = false;
+    try {
+      Thread.currentThread().setContextClassLoader(driverClassLoader);
+      HikariDataSource dataSource =
+          new DriverClassLoaderHikariDataSource(
+              createHikariConfig(configuration, jdbcUrl), driverClassLoader);
+      success = true;
+      return dataSource;
+    } finally {
+      Thread.currentThread().setContextClassLoader(previousContextClassLoader);
+      if (!success) {
+        closeDriverClassLoader(driverClassLoader);
+      }
+    }
+  }
+
+  private static HikariConfig createHikariConfig(
+      RelationalJdbcConfiguration configuration, String jdbcUrl) {
+    HikariConfig hikariConfig = new HikariConfig();
+    hikariConfig.setPoolName(POOL_NAME);
+    hikariConfig.setJdbcUrl(jdbcUrl);
+    nonBlank(configuration.driver()).ifPresent(hikariConfig::setDriverClassName);
+    nonBlank(configuration.username()).ifPresent(hikariConfig::setUsername);
+    configuration.password().ifPresent(hikariConfig::setPassword);
+    configuration.maximumPoolSize().ifPresent(hikariConfig::setMaximumPoolSize);
+    configuration.minimumIdle().ifPresent(hikariConfig::setMinimumIdle);
+    configuration.connectionTimeoutInMs().ifPresent(hikariConfig::setConnectionTimeout);
+
+    return hikariConfig;
+  }
+
+  private static Optional<URLClassLoader> createDriverClassLoader(
+      RelationalJdbcConfiguration configuration) {
+    Optional<String> configuredDriverDirectory = nonBlank(configuration.driverDirectory());
+    Optional<Path> driverDirectory =
+        configuredDriverDirectory
+            .map(Path::of)
+            .or(() -> defaultDriverDirectory().filter(Files::isDirectory));
+
+    if (driverDirectory.isEmpty()) {
+      return Optional.empty();
+    }
+
+    Path directory = driverDirectory.get();
+    if (!Files.isDirectory(directory)) {
+      throw new IllegalArgumentException(
+          "Configured JDBC driver directory does not exist: " + directory);
+    }
+
+    List<Path> driverJars = driverJars(directory);
+    if (driverJars.isEmpty()) {
+      if (configuredDriverDirectory.isPresent()) {
+        throw new IllegalArgumentException(
+            "Configured JDBC driver directory does not contain any .jar files: " + directory);
+      }
+      return Optional.empty();
+    }
+
+    LOGGER.info("Loading JDBC driver jars from {}", directory);
+    URL[] urls = driverJars.stream().map(JdbcDataSourceFactory::toUrl).toArray(URL[]::new);
+    ClassLoader parent = Thread.currentThread().getContextClassLoader();
+    if (parent == null) {
+      parent = JdbcDataSourceFactory.class.getClassLoader();
+    }
+    return Optional.of(new URLClassLoader(urls, parent));
+  }
+
+  private static Optional<Path> defaultDriverDirectory() {
+    String classPath = System.getProperty("java.class.path", "");
+    for (String classPathEntry : classPath.split(System.getProperty("path.separator"))) {
+      if (classPathEntry.isEmpty()) {
+        continue;
+      }
+      Path path = Path.of(classPathEntry);
+      Path fileName = path.getFileName();
+      if (fileName != null && "quarkus-run.jar".equals(fileName.toString())) {
+        Path parent = path.getParent();
+        return Optional.of(
+            parent == null
+                ? Path.of(DEFAULT_DRIVER_DIRECTORY)
+                : parent.resolve(DEFAULT_DRIVER_DIRECTORY));
+      }
+    }
+    return Optional.empty();
+  }
+
+  private static List<Path> driverJars(Path driverDirectory) {
+    try (Stream<Path> entries = Files.list(driverDirectory)) {
+      return entries
+          .filter(Files::isRegularFile)
+          .filter(path -> path.getFileName().toString().toLowerCase(Locale.ROOT).endsWith(".jar"))
+          .sorted(Comparator.comparing(path -> path.getFileName().toString()))
+          .toList();
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to list JDBC driver directory: " + driverDirectory, e);
+    }
+  }
+
+  private static URL toUrl(Path driverJar) {
+    try {
+      return driverJar.toUri().toURL();
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to load JDBC driver jar: " + driverJar, e);
+    }
+  }
+
+  private static void closeDriverClassLoader(URLClassLoader driverClassLoader) {
+    try {
+      driverClassLoader.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to close JDBC driver classloader", e);
+    }
+  }
+
+  private static Optional<String> nonBlank(Optional<String> value) {
+    return value.map(String::trim).filter(s -> !s.isEmpty());
+  }
+
+  private static final class DriverClassLoaderHikariDataSource extends HikariDataSource {
+    private final URLClassLoader driverClassLoader;
+
+    private DriverClassLoaderHikariDataSource(
+        HikariConfig configuration, URLClassLoader driverClassLoader) {
+      super(configuration);
+      this.driverClassLoader = driverClassLoader;
+    }
+
+    @Override
+    public void close() {
+      RuntimeException closeFailure = null;
+      try {
+        super.close();
+      } catch (RuntimeException e) {
+        closeFailure = e;
+      }
+
+      try {
+        driverClassLoader.close();
+      } catch (IOException e) {
+        RuntimeException classLoaderFailure =
+            new RuntimeException("Failed to close JDBC driver classloader", e);
+        if (closeFailure != null) {
+          closeFailure.addSuppressed(classLoaderFailure);
+        } else {
+          closeFailure = classLoaderFailure;
+        }
+      }
+
+      if (closeFailure != null) {
+        throw closeFailure;
+      }
+    }
+  }
+}

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

@@ -22,6 +22,7 @@
 
 import io.smallrye.common.annotation.Identifier;
 import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Disposes;
 import jakarta.enterprise.inject.Instance;
 import jakarta.enterprise.inject.Produces;
 import jakarta.inject.Inject;
@@ -95,7 +96,35 @@ protected JdbcMetaStoreManagerFactory() {}
   @ApplicationScoped
   static DatasourceOperations produceDatasourceOperations(
       Instance<DataSource> dataSource, RelationalJdbcConfiguration relationalJdbcConfiguration) {
-    return new DatasourceOperations(dataSource.get(), relationalJdbcConfiguration);
+    Optional<DataSource> polarisDataSource =
+        JdbcDataSourceFactory.create(relationalJdbcConfiguration);
+    return polarisDataSource
+        .map(ds -> createOwnedDatasourceOperations(ds, relationalJdbcConfiguration))
+        .orElseGet(() -> new DatasourceOperations(dataSource.get(), relationalJdbcConfiguration));
+  }
+
+  static void closeDatasourceOperations(@Disposes DatasourceOperations datasourceOperations) {
+    datasourceOperations.close();
+  }
+
+  private static DatasourceOperations createOwnedDatasourceOperations(
+      DataSource dataSource, RelationalJdbcConfiguration relationalJdbcConfiguration) {
+    try {
+      return new DatasourceOperations(dataSource, relationalJdbcConfiguration, true);
+    } catch (RuntimeException e) {
+      closeDataSource(dataSource, e);
+      throw e;
+    }
+  }
+
+  private static void closeDataSource(DataSource dataSource, RuntimeException originalException) {
+    if (dataSource instanceof AutoCloseable closeable) {
+      try {
+        closeable.close();
+      } catch (Exception closeException) {
+        originalException.addSuppressed(closeException);
+      }
+    }
   }
 
   protected PrincipalSecretsGenerator secretsGenerator(

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

@@ -35,4 +35,44 @@ public interface RelationalJdbcConfiguration {
    * the JDBC connection metadata. Supported values: "postgresql", "cockroachdb", "h2"
    */
   Optional<String> databaseType();
+
+  /** JDBC URL used when Polaris creates and owns the relational JDBC datasource. */
+  default Optional<String> jdbcUrl() {
+    return Optional.empty();
+  }
+
+  /** JDBC driver class name used when Polaris creates and owns the datasource. */
+  default Optional<String> driver() {
+    return Optional.empty();
+  }
+
+  /** Directory containing JDBC driver jars for the Polaris-managed datasource. */
+  default Optional<String> driverDirectory() {
+    return Optional.empty();
+  }
+
+  /** Database username used when Polaris creates and owns the datasource. */
+  default Optional<String> username() {
+    return Optional.empty();
+  }
+
+  /** Database password used when Polaris creates and owns the datasource. */
+  default Optional<String> password() {
+    return Optional.empty();
+  }
+
+  /** Maximum number of connections in the Polaris-managed datasource pool. */
+  default Optional<Integer> maximumPoolSize() {
+    return Optional.empty();
+  }
+
+  /** Minimum number of idle connections in the Polaris-managed datasource pool. */
+  default Optional<Integer> minimumIdle() {
+    return Optional.empty();
+  }
+
+  /** Maximum time to wait for a connection from the Polaris-managed datasource pool. */
+  default Optional<Long> connectionTimeoutInMs() {
+    return Optional.empty();
+  }
 }

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

@@ -36,11 +36,14 @@ public ProductionReadinessCheck checkRelationalJdbc(
       return ProductionReadinessCheck.OK;
     }
 
-    if (datasourceOperations.get().getDatabaseType().equals(DatabaseType.H2)) {
+    DatasourceOperations operations = datasourceOperations.get();
+    if (operations.getDatabaseType().equals(DatabaseType.H2)) {
       return ProductionReadinessCheck.of(
           ProductionReadinessCheck.Error.of(
               "The current persistence (jdbc:h2) is intended for tests only.",
-              "quarkus.datasource.jdbc.url"));
+              operations.ownsDataSource()
+                  ? "polaris.persistence.relational.jdbc.jdbc-url"
+                  : "quarkus.datasource.jdbc.url"));
     }
     return ProductionReadinessCheck.OK;
   }