This is an automated email from the ASF dual-hosted git repository.
aicam pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 078ab64d3e feat: Adding a connection pool to reduce the time spent
opening connection to Postgres (#4851)
078ab64d3e is described below
commit 078ab64d3e03cf28509c9c8e05fda558a0bcf159
Author: Matthew B. <[email protected]>
AuthorDate: Tue May 12 12:52:25 2026 -0700
feat: Adding a connection pool to reduce the time spent opening connection
to Postgres (#4851)
<!--
Thanks for sending a pull request (PR)! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
[Contributing to
Texera](https://github.com/apache/texera/blob/main/CONTRIBUTING.md)
2. Ensure you have added or run the appropriate tests for your PR
3. If the PR is work in progress, mark it a draft on GitHub.
4. Please write your PR title to summarize what this PR proposes, we
are following Conventional Commits style for PR titles as well.
5. Be sure to keep the PR description updated to reflect all changes.
-->
### What changes were proposed in this PR?
<!--
Please clarify what changes you are proposing. The purpose of this
section
is to outline the changes. Here are some tips for you:
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
3. If it is a refactoring, clarify what has been changed.
3. It would be helpful to include a before-and-after comparison using
screenshots or GIFs.
4. Please consider writing useful notes for better and faster reviews.
-->
This PR adds a HikariCP connection pool to SqlServer.scala so that jOOQ
queries borrow pre-authenticated connections from a pool instead of
opening a new TCP connection and performing SCRAM-SHA-256 authentication
on every database call.
### What changed in SqlServer.scala:
- Added a HikariConfig with maximumPoolSize=10, minimumIdle=2,
connectionTimeout=30s, idleTimeout=10min, maxLifetime=30min (chosen to
stay below typical PostgreSQL idle and load-balancer connection-reaping
windows)
- Initialized a HikariDataSource and built the jOOQ DSLContext via
DSL.using(dataSource, SQL_DIALECT) so queries draw from the pool
- Added a close() method to shut down the pool on application or test
teardown, and updated clearInstance() to call it, so test classes that
replace the singleton don't leak pool threads
- Added a documented warning against caching the DSLContext in a val or
lazy val. MockTexeraDB swaps the singleton between test classes, and a
cached context would hold a stale reference to a dead pool
### Any related issues, documentation, discussions?
<!--
Please use this section to link other resources if not mentioned
already.
1. If this PR fixes an issue, please include `Fixes #1234`, `Resolves
#1234`
or `Closes #1234`. If it is only related, simply mention the issue
number.
2. If there is design documentation, please add the link.
3. If there is a discussion in the mailing list, please add the link.
-->
closes: #4852
### How was this PR tested?
<!--
If tests were added, say they were added here. Or simply mention that if
the PR
is tested with existing test cases. Make sure to include/update test
cases that
check the changes thoroughly including negative and positive cases if
possible.
If it was tested in a way different from regular unit tests, please
clarify how
you tested step by step, ideally copy and paste-able, so that other
reviewers can
test and check, and descendants can verify in the future. If tests were
not added,
please describe why they were not added and/or why it was difficult to
add.
-->
- Existing unit and integration tests pass — the pool-backed DSLContext
is a drop-in replacement for the previous direct-driver context
- Verified the MockTexeraDB flow still works across test classes by
running the full test suite and confirming no "Connection refused"
failures from stale contexts
### Was this PR authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this
PR,
please include the phrase: 'Generated-by: ' followed by the name of the
tool
and its version. If no, write 'No'.
Please refer to the [ASF Generative Tooling
Guidance](https://www.apache.org/legal/generative-tooling.html) for
details.
-->
Co-authored with Claude opus 4.6 (Anthropic) in compliance with ASF
---------
Co-authored-by: Yicong Huang
<[email protected]>
---
access-control-service/LICENSE-binary | 1 +
amber/LICENSE-binary-java | 1 +
common/dao/build.sbt | 1 +
.../scala/org/apache/texera/dao/SqlServer.scala | 50 +++--
.../scala/org/apache/texera/dao/MockTexeraDB.scala | 1 -
.../org/apache/texera/dao/SqlServerSpec.scala | 234 +++++++++++++++++++++
computing-unit-managing-service/LICENSE-binary | 1 +
config-service/LICENSE-binary | 1 +
file-service/LICENSE-binary | 1 +
workflow-compiling-service/LICENSE-binary | 1 +
10 files changed, 273 insertions(+), 19 deletions(-)
diff --git a/access-control-service/LICENSE-binary
b/access-control-service/LICENSE-binary
index 9af04d35f5..3abc86ea44 100644
--- a/access-control-service/LICENSE-binary
+++ b/access-control-service/LICENSE-binary
@@ -247,6 +247,7 @@ Scala/Java jars:
- com.thesamet.scalapb.scalapb-runtime_2.13-0.11.20.jar
- com.typesafe.config-1.4.6.jar
- com.typesafe.scala-logging.scala-logging_2.13-3.9.5.jar
+ - com.zaxxer.HikariCP-5.1.0.jar
- io.dropwizard.dropwizard-auth-4.0.7.jar
- io.dropwizard.dropwizard-configuration-4.0.7.jar
- io.dropwizard.dropwizard-core-4.0.7.jar
diff --git a/amber/LICENSE-binary-java b/amber/LICENSE-binary-java
index 8fbd626ca4..86cdb8c8f4 100644
--- a/amber/LICENSE-binary-java
+++ b/amber/LICENSE-binary-java
@@ -291,6 +291,7 @@ Scala/Java jars:
- com.typesafe.scala-logging.scala-logging_2.13-3.9.5.jar
- com.typesafe.ssl-config-core_2.13-0.6.1.jar
- com.univocity.univocity-parsers-2.9.1.jar
+ - com.zaxxer.HikariCP-5.1.0.jar
- commons-beanutils.commons-beanutils-1.9.4.jar
- commons-cli.commons-cli-1.2.jar
- commons-codec.commons-codec-1.17.1.jar
diff --git a/common/dao/build.sbt b/common/dao/build.sbt
index b88fdbbfad..69d065e6e6 100644
--- a/common/dao/build.sbt
+++ b/common/dao/build.sbt
@@ -175,4 +175,5 @@ libraryDependencies ++= Seq(
libraryDependencies ++= Seq(
"org.postgresql" % "postgresql" % "42.7.10",
+ "com.zaxxer" % "HikariCP" % "5.1.0"
)
diff --git a/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
b/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
index 942bac4f07..6348ae41fa 100644
--- a/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
+++ b/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
@@ -19,59 +19,73 @@
package org.apache.texera.dao
+import com.zaxxer.hikari.{HikariConfig, HikariDataSource}
import org.jooq.impl.DSL
import org.jooq.{DSLContext, SQLDialect}
-import org.postgresql.ds.PGSimpleDataSource
/**
* SqlServer class that manages a connection to a PostgreSQL database using
jOOQ.
*
+ * Uses a HikariCP connection pool so that every jOOQ query borrows a
pre-authenticated
+ * connection from the pool rather than opening a new TCP + SCRAM handshake
each time.
+ *
* WARNING: Do not cache the DSLContext returned by `createDSLContext()` in a
val or lazy val.
* During testing, `MockTexeraDB` replaces the SqlServer instance between
test classes.
* A cached DSLContext will hold a stale reference to a dead database
connection from a previous test class,
* causing "Connection refused" errors when tests run together.
* Use `def` to ensure the connection is looked up each time.
*
- * @param url The database connection URL.
- * @param user The username for authenticating with the database.
+ * @param url The JDBC connection URL.
+ * @param user The username for authenticating with the database.
* @param password The password for authenticating with the database.
*/
class SqlServer private (url: String, user: String, password: String) {
val SQL_DIALECT: SQLDialect = SQLDialect.POSTGRES
- private val dataSource: PGSimpleDataSource = new PGSimpleDataSource()
- var context: DSLContext = {
- dataSource.setUrl(url)
- dataSource.setUser(user)
- dataSource.setPassword(password)
- dataSource.setConnectTimeout(5)
- DSL.using(dataSource, SQL_DIALECT)
+
+ private val hikariConfig: HikariConfig = {
+ val cfg = new HikariConfig()
+ cfg.setJdbcUrl(url)
+ cfg.setUsername(user)
+ cfg.setPassword(password)
+ cfg.setPoolName("texera-hikari")
+ cfg.setMaximumPoolSize(10)
+ cfg.setMinimumIdle(2)
+ // How long a caller waits for a connection before throwing (ms)
+ cfg.setConnectionTimeout(30000)
+ // How long an idle connection stays in the pool before being retired (ms)
+ cfg.setIdleTimeout(600000)
+ // Maximum lifetime of any connection in the pool (ms); must be <
PostgreSQL's idle timeout
+ cfg.setMaxLifetime(1800000)
+ cfg
}
+ private val dataSource: HikariDataSource = new HikariDataSource(hikariConfig)
+
+ var context: DSLContext = DSL.using(dataSource, SQL_DIALECT)
+
def createDSLContext(): DSLContext = context
def replaceDSLContext(newContext: DSLContext): Unit = {
context = newContext
}
+
+ def close(): Unit = {
+ if (!dataSource.isClosed) dataSource.close()
+ }
}
object SqlServer {
private var instance: Option[SqlServer] = None
def initConnection(url: String, user: String, password: String): Unit = {
- if (instance.isEmpty) {
- val server = new SqlServer(url, user, password)
- instance = Some(server)
- }
+ instance.foreach(_.close())
+ instance = Some(new SqlServer(url, user, password))
}
def getInstance(): SqlServer = {
instance.get
}
- def clearInstance(): Unit = {
- instance = None
- }
-
/**
* A utility function for create a transaction block using given sql context
* @param dsl the sql context
diff --git a/common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala
b/common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala
index e13ff696cf..3ae97b62e0 100644
--- a/common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala
+++ b/common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala
@@ -67,7 +67,6 @@ trait MockTexeraDB {
value.close()
dbInstance = None
dslContext = None
- SqlServer.clearInstance()
case None =>
// do nothing
}
diff --git
a/common/dao/src/test/scala/org/apache/texera/dao/SqlServerSpec.scala
b/common/dao/src/test/scala/org/apache/texera/dao/SqlServerSpec.scala
new file mode 100644
index 0000000000..0e786b6bb7
--- /dev/null
+++ b/common/dao/src/test/scala/org/apache/texera/dao/SqlServerSpec.scala
@@ -0,0 +1,234 @@
+/*
+ * 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.texera.dao
+
+import com.zaxxer.hikari.{HikariConfig, HikariDataSource}
+import org.jooq.impl.DSL
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll}
+
+class SqlServerSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll
with MockTexeraDB {
+
+ override def beforeAll(): Unit = initializeDBAndReplaceDSLContext()
+ override def afterAll(): Unit = shutdownDB()
+
+ // -------------------------------------------------------------------------
+ // SqlServer.withTransaction
+ //
+ // getDSLContext is backed by the embedded Postgres DataSource, so each
+ // top-level query borrows a connection from the pool. withTransaction
+ // binds a single connection for the duration of the block, making rollback
+ // and commit behaviour fully observable.
+ // -------------------------------------------------------------------------
+
+ "SqlServer.withTransaction" should "return the value produced by the block"
in {
+ val result = SqlServer.withTransaction(getDSLContext) { _ => 42 }
+ result shouldBe 42
+ }
+
+ it should "commit the block's work so subsequent queries observe the
changes" in {
+ // SELECT 1 is a lightweight live query; completing without error confirms
+ // the transaction committed and the connection was returned cleanly.
+ val result = SqlServer.withTransaction(getDSLContext) { ctx =>
+ ctx.selectOne().fetchOne().value1()
+ }
+ result shouldBe 1
+ }
+
+ it should "re-throw the exception when the block throws" in {
+ val boom = new RuntimeException("intentional failure")
+ val thrown = intercept[RuntimeException] {
+ SqlServer.withTransaction(getDSLContext) { _ => throw boom }
+ }
+ thrown.getMessage should include("intentional failure")
+ }
+
+ it should "roll back all DML in the block when an exception is thrown" in {
+ // A permanent (non-TEMP) table is used so every connection from the pool
+ // can see it; TEMP tables are session-scoped and would be invisible across
+ // pool connections.
+ val dsl = getDSLContext
+ dsl.execute("CREATE TABLE IF NOT EXISTS _txn_rollback_test (v INT)")
+ try {
+ intercept[RuntimeException] {
+ SqlServer.withTransaction(dsl) { ctx =>
+ ctx.execute("INSERT INTO _txn_rollback_test VALUES (99)")
+ throw new RuntimeException("force rollback")
+ }
+ }
+ // The INSERT was inside the rolled-back transaction, so the table must
+ // still be empty.
+ dsl.fetchCount(DSL.table(DSL.name("_txn_rollback_test"))) shouldBe 0
+ } finally {
+ dsl.execute("DROP TABLE IF EXISTS _txn_rollback_test")
+ }
+ }
+
+ it should "support nested return types beyond Int" in {
+ val result = SqlServer.withTransaction(getDSLContext) { ctx =>
+ ctx.selectOne().fetchOne().value1().toString
+ }
+ result shouldBe "1"
+ }
+
+ // -------------------------------------------------------------------------
+ // HikariCP pool lifecycle and configuration
+ //
+ // These tests create their own HikariDataSource against the embedded
Postgres
+ // instance so they can drive the pool directly, independently of the
+ // DSLContext replacement that MockTexeraDB applies for its own queries.
+ // -------------------------------------------------------------------------
+
+ private def buildPool(
+ maxSize: Int = 5,
+ minIdle: Int = 1,
+ poolName: String = "spec-pool"
+ ): HikariDataSource = {
+ // Use the default "postgres" database so no schema setup is needed.
+ val jdbcUrl = getDBInstance.getJdbcUrl("postgres", "postgres")
+ val cfg = new HikariConfig()
+ cfg.setJdbcUrl(jdbcUrl)
+ cfg.setUsername("postgres")
+ cfg.setPassword("")
+ cfg.setPoolName(poolName)
+ cfg.setMaximumPoolSize(maxSize)
+ cfg.setMinimumIdle(minIdle)
+ cfg.setConnectionTimeout(5000)
+ new HikariDataSource(cfg)
+ }
+
+ "HikariCP pool" should "provide a usable connection that can execute
queries" in {
+ val ds = buildPool()
+ try {
+ val conn = ds.getConnection
+ try {
+ val rs = conn.prepareStatement("SELECT 1").executeQuery()
+ rs.next() shouldBe true
+ rs.getInt(1) shouldBe 1
+ } finally conn.close()
+ } finally ds.close()
+ }
+
+ it should "apply the configured pool name" in {
+ val ds = buildPool(poolName = "my-named-pool")
+ try {
+ ds.getHikariConfigMXBean.getPoolName shouldBe "my-named-pool"
+ } finally ds.close()
+ }
+
+ it should "apply the configured maximum pool size" in {
+ val ds = buildPool(maxSize = 7)
+ try {
+ ds.getHikariConfigMXBean.getMaximumPoolSize shouldBe 7
+ } finally ds.close()
+ }
+
+ it should "apply the configured minimum idle connections" in {
+ val ds = buildPool(minIdle = 2)
+ try {
+ ds.getHikariConfigMXBean.getMinimumIdle shouldBe 2
+ } finally ds.close()
+ }
+
+ it should "count a borrowed connection as active" in {
+ val ds = buildPool()
+ try {
+ val conn = ds.getConnection
+ try {
+ ds.getHikariPoolMXBean.getActiveConnections should be >= 1
+ } finally conn.close()
+ } finally ds.close()
+ }
+
+ it should "decrement active count and increment idle count once a connection
is returned" in {
+ val ds = buildPool()
+ try {
+ val conn = ds.getConnection
+ conn.close()
+ ds.getHikariPoolMXBean.getActiveConnections shouldBe 0
+ ds.getHikariPoolMXBean.getIdleConnections should be >= 1
+ } finally ds.close()
+ }
+
+ it should "allow up to the maximum pool size connections to be borrowed
concurrently" in {
+ val ds = buildPool(maxSize = 3)
+ try {
+ val c1 = ds.getConnection
+ val c2 = ds.getConnection
+ val c3 = ds.getConnection
+ ds.getHikariPoolMXBean.getActiveConnections shouldBe 3
+ c1.close(); c2.close(); c3.close()
+ } finally ds.close()
+ }
+
+ it should "report isClosed as false while open and true after close" in {
+ val ds = buildPool()
+ ds.isClosed shouldBe false
+ ds.close()
+ ds.isClosed shouldBe true
+ }
+
+ it should "reject getConnection after the pool has been closed" in {
+ val ds = buildPool()
+ ds.close()
+ // HikariCP throws an SQLException (wrapped as RuntimeException by the
pool)
+ // when a caller tries to borrow from a closed pool.
+ assertThrows[Exception](ds.getConnection)
+ }
+
+ // -------------------------------------------------------------------------
+ // SqlServer.close()
+ //
+ // The instance's private HikariDataSource is the only resource that needs
+ // explicit release; close() guards it against double-close. These tests
+ // construct a fresh SqlServer via initConnection (the only public entry
+ // point — the class constructor is private) and assert against the
+ // underlying pool via reflection, which avoids broadening the class API
+ // just to make this branch observable.
+ // -------------------------------------------------------------------------
+
+ private def datasourceOf(instance: SqlServer): HikariDataSource = {
+ val field = classOf[SqlServer].getDeclaredField("dataSource")
+ field.setAccessible(true)
+ field.get(instance).asInstanceOf[HikariDataSource]
+ }
+
+ "SqlServer.close" should "shut down the underlying HikariDataSource and be
idempotent" in {
+ val jdbcUrl = getDBInstance.getJdbcUrl("postgres", "postgres")
+ // Replaces the singleton — initConnection internally calls close() on the
+ // prior instance, which is itself an exercise of the same path. The trait
+ // holds its own DSLContext separately, so other tests' database access is
+ // unaffected by this replacement.
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
+ val instance = SqlServer.getInstance()
+ val ds = datasourceOf(instance)
+
+ ds.isClosed shouldBe false
+ instance.close()
+ ds.isClosed shouldBe true
+
+ // Second close() must take the `dataSource.isClosed` branch and return
+ // without throwing. Calling Hikari's close() twice would itself be safe
+ // today, but the guard is what this assertion pins.
+ noException should be thrownBy instance.close()
+ ds.isClosed shouldBe true
+ }
+}
diff --git a/computing-unit-managing-service/LICENSE-binary
b/computing-unit-managing-service/LICENSE-binary
index 229fee7fde..37d78afb47 100644
--- a/computing-unit-managing-service/LICENSE-binary
+++ b/computing-unit-managing-service/LICENSE-binary
@@ -274,6 +274,7 @@ Scala/Java jars:
- com.typesafe.play.play-functional_2.13-2.10.6.jar
- com.typesafe.play.play-json_2.13-2.10.6.jar
- com.typesafe.scala-logging.scala-logging_2.13-3.9.5.jar
+ - com.zaxxer.HikariCP-5.1.0.jar
- commons-beanutils.commons-beanutils-1.9.4.jar
- commons-cli.commons-cli-1.2.jar
- commons-codec.commons-codec-1.17.1.jar
diff --git a/config-service/LICENSE-binary b/config-service/LICENSE-binary
index 73bff916d9..78c3b7878f 100644
--- a/config-service/LICENSE-binary
+++ b/config-service/LICENSE-binary
@@ -247,6 +247,7 @@ Scala/Java jars:
- com.thesamet.scalapb.scalapb-runtime_2.13-0.11.20.jar
- com.typesafe.config-1.4.6.jar
- com.typesafe.scala-logging.scala-logging_2.13-3.9.5.jar
+ - com.zaxxer.HikariCP-5.1.0.jar
- io.dropwizard.dropwizard-auth-4.0.7.jar
- io.dropwizard.dropwizard-configuration-4.0.7.jar
- io.dropwizard.dropwizard-core-4.0.7.jar
diff --git a/file-service/LICENSE-binary b/file-service/LICENSE-binary
index 01172fafb8..138fd6cad0 100644
--- a/file-service/LICENSE-binary
+++ b/file-service/LICENSE-binary
@@ -268,6 +268,7 @@ Scala/Java jars:
- com.thesamet.scalapb.scalapb-runtime_2.13-0.11.20.jar
- com.typesafe.config-1.4.6.jar
- com.typesafe.scala-logging.scala-logging_2.13-3.9.5.jar
+ - com.zaxxer.HikariCP-5.1.0.jar
- commons-beanutils.commons-beanutils-1.9.4.jar
- commons-cli.commons-cli-1.2.jar
- commons-codec.commons-codec-1.17.1.jar
diff --git a/workflow-compiling-service/LICENSE-binary
b/workflow-compiling-service/LICENSE-binary
index 7620977853..5b7548a4ed 100644
--- a/workflow-compiling-service/LICENSE-binary
+++ b/workflow-compiling-service/LICENSE-binary
@@ -269,6 +269,7 @@ Scala/Java jars:
- com.thesamet.scalapb.scalapb-runtime_2.13-0.11.20.jar
- com.typesafe.config-1.4.6.jar
- com.typesafe.scala-logging.scala-logging_2.13-3.9.5.jar
+ - com.zaxxer.HikariCP-5.1.0.jar
- com.univocity.univocity-parsers-2.9.1.jar
- commons-beanutils.commons-beanutils-1.9.4.jar
- commons-cli.commons-cli-1.2.jar
