This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6238-a7989ee78e27ab7b909bdf8aaed586a5f030528c in repository https://gitbox.apache.org/repos/asf/texera.git
commit cb5f2f5775edf8d70b8ceae2ecea1c3a05d5e9b5 Author: Neil Ketteringham <[email protected]> AuthorDate: Mon Jul 27 13:26:07 2026 -0700 refactor: Refactor MockTexeraDB into a JVM singleton (#6238) ### What changes were proposed in this PR? Closes #6063 by completely refactoring the `MockTexeraDB` infrastructure to transition from a distributed process architecture to a centralized, shared-singleton architecture. 1. **High-Performance Shared Singleton Architecture:** Refactored the core setup by splitting the original `MockTexeraDB` trait into a companion `object MockTexeraDB` singleton and a lightweight `trait MockTexeraDB` interface wrapper. The background `EmbeddedPostgres` database engine process is now booted **exactly once** per JVM lifecycle via synchronized initialization. Individual test suites dynamically provision an isolated schema sandbox via cheap `CREATE DATABASE` queries instead of spawning heavy process-level instances. 2. **DDL Caching and Regex Optimization:** Removed massive runtime file I/O and processing bottlenecks. The full-text search index string regex modifications and SQL script cleaning are handled once and cached globally in memory (`ddlScript`), preventing parallel test suites from thrashing disk resources over the same `sql/texera_ddl.sql` file. 3. **Per-Suite HikariCP Connection Pooling:** Each isolated schema sandbox is now backed by a real HikariCP pool (bounded maximumPoolSize) rather than a single shared connection. This gives concurrent transactions distinct connections — matching production — so they no longer trample one connection's autoCommit flag. This pooling is the mechanism that lets the high-concurrency specs (e.g. DatasetResourceSpec) converge without deadlocks or connection-state leakage. 4. **Teardown API (shutdownDB → closeConnectionPool):** Because the trait now owns explicit pool resources, the teardown method was renamed to closeConnectionPool() and all in-tree callers updated. To preserve backward compatibility, shutdownDB() is retained as a thin alias (def shutdownDB(): Unit = closeConnectionPool()) so existing/incoming specs on main continue to compile unchanged. 5. **E2E specs moved onto embedded Postgres:** TestUtils.initiateTexeraDBForTestCases now provisions its isolated database on MockTexeraDB's embedded Postgres instead of an external test Postgres instance. This depends on #4179. The helper was also hardened — the DDL connection is now closed via Using.resource (no leak on failure) and a dead DROP DATABASE IF EXISTS on the freshly generated UUID name was removed. ### Performance Impact Moving database instantiation, parsing, and regex evaluation out of the distributed trait lifecycle and into a centralized static memory layer resulted in a dramatic reduction in test suite runtime. * **Main Branch Baseline / Prior Implementation:** 5 minutes 40 seconds * **This PR:** **4 minutes 5 seconds** (Total execution time reduced by ~28% locally). ### Any related issues, documentation, discussions? Follow up on #4525 where a refactor of `MockTexeraDB` was needed to allow for future concurrent execution of test suites. Structural baseline was cherry-picked and heavily adapted from #4527, specifically commit `bd93161` written by @Yicong-Huang. ### How was this PR tested? Verified by running full, clean compilations and local stress tests on JDK 17 to validate that the structural split preserves total API backward compatibility across the codebase. Additionally validated that high-concurrency race conditions (such as the asynchronous multi-threaded tests inside `DatasetResourceSpec`) converge gracefully without experiencing deadlocks, thread crashes, or connection state leakage. All 1,600+ test suite assertions across the system are green. Run configuration command used to validate the project modules: ```bash sbt clean compile test ``` ### Was this PR authored or co-authored using generative AI tooling? Yes. Generative AI (Gemini) was utilized as an external assistant to analyze complex multi-threaded JDBC/jOOQ connection trace errors and help structure the safe connection lifecycle fallback logic in SqlServer.scala. No direct IDE-integrated automation tools were used to auto-generate codebase files. --------- Signed-off-by: Neil Ketteringham <[email protected]> Co-authored-by: Yicong Huang <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../apache/texera/AccessControlResourceSpec.scala | 2 +- .../scheduling/DefaultCostEstimatorSpec.scala | 2 +- .../apache/texera/amber/engine/e2e/TestUtils.scala | 28 ++- .../texera/web/resource/FeedbackResourceSpec.scala | 2 +- .../dashboard/file/WorkflowResourceSpec.scala | 2 +- .../user/project/ProjectAccessResourceSpec.scala | 2 +- .../user/workflow/WorkflowAccessResourceSpec.scala | 2 +- .../workflow/WorkflowExecutionsResourceSpec.scala | 2 +- .../workflow/WorkflowVersionResourceSpec.scala | 2 +- .../pythonvirtualenvironment/PveResourceSpec.scala | 2 +- .../ExecutionsMetadataPersistServiceSpec.scala | 2 +- .../texera/auth/util/ComputingUnitAccessSpec.scala | 2 +- .../scala/org/apache/texera/dao/MockTexeraDB.scala | 270 ++++++++++++--------- .../org/apache/texera/dao/SqlServerSpec.scala | 2 +- .../texera/amber/storage/FileResolverSpec.scala | 2 +- .../service/resource/DatasetResourceSpec.scala | 33 ++- .../service/util/StagedFileCleanupJobSpec.scala | 2 +- .../resource/NotebookMigrationResourceSpec.scala | 2 +- 18 files changed, 204 insertions(+), 157 deletions(-) diff --git a/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala b/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala index 365f5f885f..3677e8373d 100644 --- a/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala +++ b/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala @@ -144,7 +144,7 @@ class AccessControlResourceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } "AccessControlResource" should "return FORBIDDEN for a GET request without a token" in { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala index b8f5617e13..595a9b07c9 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala @@ -116,7 +116,7 @@ class DefaultCostEstimatorSpec override protected def afterEach(): Unit = { document.clear() - shutdownDB() + closeConnectionPool() } "DefaultCostEstimator" should "use fallback method when no past statistics are available" in { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala index 399b351342..ac3bf167d3 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala @@ -21,7 +21,6 @@ package org.apache.texera.amber.engine.e2e import com.twitter.util.{Await, Duration, Promise, Return, Throw, Try} import org.apache.pekko.actor.ActorSystem -import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.core.executor.OpExecInitInfo import org.apache.texera.amber.core.storage.DocumentFactory import org.apache.texera.amber.core.storage.model.VirtualDocument @@ -224,12 +223,33 @@ object TestUtils { * If a test case accesses the user system through singleton resources that cache the DSLContext (e.g., executes a * workflow, which accesses WorkflowExecutionsResource), we use a separate texera_db specifically for such test cases. * Note such test cases need to clean up the database at the end of running each test case. + * + * This backs the e2e specs with MockTexeraDB's embedded Postgres instead of an external test Postgres + * (depends on #4179). */ def initiateTexeraDBForTestCases(): Unit = { + org.apache.texera.dao.MockTexeraDB.ensureInitialized() + val embedded = org.apache.texera.dao.MockTexeraDB.getDBInstance + + val dbName = "texera_db_for_test_cases_" + java.util.UUID.randomUUID().toString.replace("-", "") + + scala.util.Using.resource(embedded.getPostgresDatabase.getConnection) { conn => + scala.util.Using.resource(conn.createStatement()) { stmt => + stmt.execute(s"CREATE DATABASE $dbName") + } + } + + scala.util.Using.resource(embedded.getDatabase("postgres", dbName).getConnection) { + targetDbConn => + scala.util.Using.resource(targetDbConn.createStatement()) { stmt => + stmt.execute(org.apache.texera.dao.MockTexeraDB.getDDLScript) + } + } + SqlServer.initConnection( - StorageConfig.jdbcUrlForTestCases, - StorageConfig.jdbcUsername, - StorageConfig.jdbcPassword + embedded.getJdbcUrl("postgres", dbName), + "postgres", + "" ) } diff --git a/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala index 9b081f7efa..10e1e5b357 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala @@ -66,7 +66,7 @@ class FeedbackResourceSpec otherSessionUser = new SessionUser(otherUser) } - override protected def afterAll(): Unit = shutdownDB() + override protected def afterAll(): Unit = closeConnectionPool() private def clearFeedback(): Unit = { getDSLContext.deleteFrom(FEEDBACK).execute() diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala index 7e6f9572d3..d41b452f6b 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala @@ -193,7 +193,7 @@ class WorkflowResourceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } private def getKeywordsArray(keywords: String*): util.ArrayList[String] = { diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala index 185d5afcd1..2100be4b42 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala @@ -66,7 +66,7 @@ class ProjectAccessResourceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } private def createUser(uid: Int, name: String, email: String): User = { diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala index 163f7b2683..72237b8bef 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala @@ -178,7 +178,7 @@ class WorkflowAccessResourceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } "WorkflowAccessResource.revokeAccess" should "successfully revoke access when user has WRITE permission" in { diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala index c01489aa86..fdda487953 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala @@ -185,7 +185,7 @@ class WorkflowExecutionsResourceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } // ─── helpers ────────────────────────────────────────────────────────────── diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala index ecf704f663..3e86f38c2e 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala @@ -83,7 +83,7 @@ class WorkflowVersionResourceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } private def createWorkflowContent(value: String): String = { diff --git a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala index 416bd07a74..9904e3eb35 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala @@ -116,7 +116,7 @@ class PveResourceSpec override protected def afterAll(): Unit = { PveManager.runProcess = realRunner - shutdownDB() + closeConnectionPool() } override protected def beforeEach(): Unit = { diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala index 1d67fa0331..ed67b78cc5 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala @@ -72,7 +72,7 @@ class ExecutionsMetadataPersistServiceSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } override protected def beforeEach(): Unit = { diff --git a/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala b/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala index 6a5c519fc0..d6589ba4d0 100644 --- a/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala +++ b/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala @@ -81,7 +81,7 @@ class ComputingUnitAccessSpec } } - override def afterAll(): Unit = shutdownDB() + override def afterAll(): Unit = closeConnectionPool() "getComputingUnitAccess" should "return NONE when the computing unit does not exist" in { ComputingUnitAccess.getComputingUnitAccess(999, 1) shouldBe PrivilegeEnum.NONE 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 3ae97b62e0..86449e0a11 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 @@ -19,146 +19,178 @@ package org.apache.texera.dao +import com.zaxxer.hikari.{HikariConfig, HikariDataSource} import io.zonky.test.db.postgres.embedded.EmbeddedPostgres -import org.jooq.impl.DSL +import org.apache.texera.dao.MockTexeraDB.{MaxPoolSize, password, username} +import org.jooq.impl.{DSL, DataSourceConnectionProvider, DefaultConfiguration} import org.jooq.{DSLContext, SQLDialect} +import org.scalatest.{Outcome, TestSuite, TestSuiteMixin} import java.nio.file.Paths -import java.sql.{Connection, DriverManager} +import java.sql.DriverManager import scala.io.Source +import scala.util.Using -trait MockTexeraDB { - - private var dbInstance: Option[EmbeddedPostgres] = None - private var dslContext: Option[DSLContext] = None - private val database: String = "texera_db" +/** + * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that mix + * in this trait share one Postgres instance for the lifetime of the JVM. + */ +object MockTexeraDB { private val username: String = "postgres" private val password: String = "" + private val texeraDDLPath = "sql/texera_ddl.sql" + private val splitDatabaseRegex = "(?m)^CREATE DATABASE :\"DB_NAME\";" - def executeScriptInJDBC(conn: Connection, script: String): Unit = { - assert(dbInstance.nonEmpty) - conn.prepareStatement(script).execute() - conn.close() - } + val MaxPoolSize: Int = math.max(10, Runtime.getRuntime.availableProcessors() * 2) - def getDSLContext: DSLContext = { - dslContext match { - case Some(value) => value - case None => - throw new RuntimeException( - "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?" - ) - } - } + @volatile private var dbInstance: Option[EmbeddedPostgres] = None + @volatile private var ddlScript: Option[String] = None + + def ensureInitialized(): Unit = + synchronized { + if (dbInstance.isDefined && ddlScript.isDefined) return + + if (dbInstance.isEmpty) { + val driver = new org.postgresql.Driver() + DriverManager.registerDriver(driver) - def getDBInstance: EmbeddedPostgres = { - dbInstance match { - case Some(value) => value - case None => - throw new RuntimeException( - "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?" + // Boot the heavy JVM engine exactly once + dbInstance = Some(EmbeddedPostgres.builder().start()) + } + + val ddlPath = Paths.get(texeraDDLPath).toRealPath() + val source = Source.fromFile(ddlPath.toString) + val content = + try source.mkString + finally source.close() + + val parts: Array[String] = content.split(splitDatabaseRegex) + val sqlBody = parts + .lift(1) + .getOrElse( + throw new RuntimeException( + s"Couldn't split SQL body from $texeraDDLPath: " + + s"expected it to match pattern $splitDatabaseRegex" + ) ) - } - } - def shutdownDB(): Unit = { - dbInstance match { - case Some(value) => - value.close() - dbInstance = None - dslContext = None - case None => - // do nothing + def removeCCommands(sql: String): String = + sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n") + + val tablesAndIndexCreation = removeCCommands(sqlBody) + + val blockPattern = + """(?s)-- START Fulltext search index creation \(DO NOT EDIT THIS LINE\).*?-- END Fulltext search index creation \(DO NOT EDIT THIS LINE\)\n?""".r + val replacementText = + """CREATE INDEX idx_workflow_name_description_content ON workflow USING GIN (to_tsvector('english', COALESCE(name, '') || ' ' || COALESCE(description, '') || ' ' || COALESCE(content, ''))); + |CREATE INDEX idx_user_name ON "user" USING GIN (to_tsvector('english', COALESCE(name, ''))); + |CREATE INDEX idx_user_project_name_description ON project USING GIN (to_tsvector('english', COALESCE(name, '') || ' ' || COALESCE(description, ''))); + |CREATE INDEX idx_dataset_name_description ON dataset USING GIN (to_tsvector('english', COALESCE(name, '') || ' ' || COALESCE(description, ''))); + |CREATE INDEX idx_dataset_version_name ON dataset_version USING GIN (to_tsvector('english', COALESCE(name, '')));""".stripMargin + + // Cache the cleaned script so parallel suites don't have to re-read the file + ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation, replacementText).trim) } - } - def initializeDBAndReplaceDSLContext(): Unit = { - assert(dbInstance.isEmpty && dslContext.isEmpty) + def getDBInstance: EmbeddedPostgres = + dbInstance.getOrElse(throw new RuntimeException("DB not initialized")) + def getDDLScript: String = ddlScript.getOrElse(throw new RuntimeException("DDL not loaded")) +} - val driver = new org.postgresql.Driver() - DriverManager.registerDriver(driver) +trait MockTexeraDB extends TestSuiteMixin { this: TestSuite => + private var testScopedContext: Option[DSLContext] = None + protected var dataSource: Option[HikariDataSource] = None + protected var uniqueDbName: String = "" + + def createHikariConfig(jbdcUrl: String): HikariConfig = { + val hikariConfig = new HikariConfig() + hikariConfig.setJdbcUrl(jbdcUrl) + hikariConfig.setUsername(username) + hikariConfig.setPassword(password) + hikariConfig.setMaximumPoolSize(MaxPoolSize) + hikariConfig + } - val embedded = EmbeddedPostgres.builder().start() + def initializeDBAndReplaceDSLContext(): Unit = + synchronized { + if (dataSource.isEmpty || dataSource.get.isClosed) { + MockTexeraDB.ensureInitialized() + val embedded = MockTexeraDB.getDBInstance + + uniqueDbName = "texera_db_" + java.util.UUID.randomUUID().toString.replace("-", "") + Using.resource(embedded.getPostgresDatabase.getConnection) { defaultConn => + Using.resource(defaultConn.createStatement()) { stmt => + stmt.execute(s"CREATE DATABASE $uniqueDbName") + } + } + + // Run the DDL once via a throwaway connection (autoCommit is TRUE by default, + // so the schema is permanently committed to this suite's isolated database). + Using.resource(embedded.getDatabase("postgres", uniqueDbName).getConnection) { conn => + Using.resource(conn.createStatement()) { stmt => + stmt.execute(MockTexeraDB.getDDLScript) + } + } + + val jdbcUrl = embedded.getJdbcUrl("postgres", uniqueDbName) + val ds = new HikariDataSource(createHikariConfig(jbdcUrl = jdbcUrl)) + dataSource = Some(ds) + + val jooqCfg = new DefaultConfiguration() + jooqCfg.set(new DataSourceConnectionProvider(ds)) + jooqCfg.set(SQLDialect.POSTGRES) + val scopedCtx = DSL.using(jooqCfg) + testScopedContext = Some(scopedCtx) + + SqlServer.initConnection(jdbcUrl, username, password) + SqlServer.getInstance().replaceDSLContext(scopedCtx) + } + } - dbInstance = Some(embedded) + // NOTE: This shared JVM singleton design assumes test suites run sequentially. + // If parallel suite execution is enabled in the future (#4525), thread safety + // and schema isolation must be re-evaluated. + abstract override def withFixture(test: NoArgTest): Outcome = { + initializeDBAndReplaceDSLContext() - val ddlPath = { - Paths.get("sql/texera_ddl.sql").toRealPath() + val sqlServerInstance = SqlServer.getInstance() + val activeContext = testScopedContext.get + + try { + sqlServerInstance.replaceDSLContext(activeContext) + super.withFixture(test) + } finally { + /*TODO: Need to truncate texeraDB tables when the fixture is complete. + This will require refactoring the spec tests using MockTexeraDB + to move initialization logic outside of BeforeAll into BeforeEach + */ } - val source = Source.fromFile(ddlPath.toString) - val content = - try { - source.mkString - } finally { - source.close() + } + + def getDSLContext: DSLContext = + synchronized { + if (testScopedContext.isEmpty) { + initializeDBAndReplaceDSLContext() } - val parts: Array[String] = content.split("(?m)^CREATE DATABASE :\"DB_NAME\";") - def removeCCommands(sql: String): String = - sql.linesIterator - .filterNot(_.trim.startsWith("\\c")) - .mkString("\n") - val createDBStatement = - """DROP DATABASE IF EXISTS texera_db; - |CREATE DATABASE texera_db;""".stripMargin - executeScriptInJDBC(embedded.getPostgresDatabase.getConnection, createDBStatement) - val texeraDB = embedded.getDatabase(username, database) - var tablesAndIndexCreation = removeCCommands(parts(1)) - - // remove indexes creation for pgroonga because we cannot install the plugin - val blockPattern = - """(?s)-- START Fulltext search index creation \(DO NOT EDIT THIS LINE\).*?-- END Fulltext search index creation \(DO NOT EDIT THIS LINE\)\n?""".r - // replace with native fulltext indexes - val replacementText = - """CREATE INDEX idx_workflow_name_description_content - | ON workflow - | USING GIN ( - | to_tsvector('english', - | COALESCE(name, '') || ' ' || - | COALESCE(description, '') || ' ' || - | COALESCE(content, '') - | ) - | ); - | - |CREATE INDEX idx_user_name - | ON "user" - | USING GIN ( - | to_tsvector('english', - | COALESCE(name, '') - | ) - | ); - | - |CREATE INDEX idx_user_project_name_description - | ON project - | USING GIN ( - | to_tsvector('english', - | COALESCE(name, '') || ' ' || - | COALESCE(description, '') - | ) - | ); - | - |CREATE INDEX idx_dataset_name_description - | ON dataset - | USING GIN ( - | to_tsvector('english', - | COALESCE(name, '') || ' ' || - | COALESCE(description, '') - | ) - | ); - | - |CREATE INDEX idx_dataset_version_name - | ON dataset_version - | USING GIN ( - | to_tsvector('english', - | COALESCE(name, '') - | ) - | );""".stripMargin - - tablesAndIndexCreation = blockPattern.replaceAllIn(tablesAndIndexCreation, replacementText).trim - executeScriptInJDBC(texeraDB.getConnection, tablesAndIndexCreation) - SqlServer.initConnection(embedded.getJdbcUrl(username, database), username, password) - val sqlServerInstance = SqlServer.getInstance() - dslContext = Some(DSL.using(texeraDB, SQLDialect.POSTGRES)) + testScopedContext.get + } - sqlServerInstance.replaceDSLContext(dslContext.get) + def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance + def newRawConnection(): java.sql.Connection = { + MockTexeraDB.getDBInstance.getDatabase(username, uniqueDbName).getConnection } + + def closeConnectionPool(): Unit = { + //git issue #6063 asked for a no-op shutdown, however this was before + //MockTexeraDB held a connection pool, since we own explicit resources + //we must close them. + synchronized { + try dataSource.foreach(ds => if (!ds.isClosed) ds.close()) + catch { case e: Exception => e.printStackTrace() } + finally { dataSource = None; testScopedContext = None } + } + } + + /** Backwards-compat alias for specs on `main` that still call shutdownDB(). */ + def shutdownDB(): Unit = closeConnectionPool() } 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 index 0e786b6bb7..22c2350f44 100644 --- a/common/dao/src/test/scala/org/apache/texera/dao/SqlServerSpec.scala +++ b/common/dao/src/test/scala/org/apache/texera/dao/SqlServerSpec.scala @@ -28,7 +28,7 @@ import org.scalatest.{BeforeAndAfterAll} class SqlServerSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll with MockTexeraDB { override def beforeAll(): Unit = initializeDBAndReplaceDSLContext() - override def afterAll(): Unit = shutdownDB() + override def afterAll(): Unit = closeConnectionPool() // ------------------------------------------------------------------------- // SqlServer.withTransaction diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala index 62daa811d3..6916ec6641 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala @@ -163,7 +163,7 @@ class FileResolverSpec } override protected def afterAll(): Unit = { - shutdownDB() + closeConnectionPool() } } diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala index 99d1fd8311..fb27fa6731 100644 --- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala @@ -269,7 +269,7 @@ class DatasetResourceSpec } override protected def afterAll(): Unit = { - try shutdownDB() + try closeConnectionPool() finally { try savedLevels.foreach { case (name, prev) => setLoggerLevel(name, prev) } finally super .afterAll() @@ -1590,8 +1590,7 @@ class DatasetResourceSpec val filePath = uniqueFilePath("init-session-row-locked") initUpload(filePath, numParts = 2).getStatus shouldEqual 200 - val connectionProvider = getDSLContext.configuration().connectionProvider() - val connection = connectionProvider.acquire() + val connection = newRawConnection() connection.setAutoCommit(false) try { @@ -1613,7 +1612,7 @@ class DatasetResourceSpec ex.getResponse.getStatus shouldEqual 409 } finally { connection.rollback() - connectionProvider.release(connection) + connection.close() } // lock released => init works again @@ -1979,8 +1978,8 @@ class DatasetResourceSpec val filePath = uniqueFilePath("init-lock-409") initUpload(filePath, numParts = 2).getStatus shouldEqual 200 - val connectionProvider = getDSLContext.configuration().connectionProvider() - val connection = connectionProvider.acquire() + // Open a completely independent connection to simulate a second concurrent user + val connection = newRawConnection() connection.setAutoCommit(false) try { @@ -2002,7 +2001,7 @@ class DatasetResourceSpec ex.getResponse.getStatus shouldEqual 409 } finally { connection.rollback() - connectionProvider.release(connection) + connection.close() } } @@ -2222,8 +2221,7 @@ class DatasetResourceSpec initUpload(filePath, numParts = 2) val uploadId = fetchUploadIdOrFail(filePath) - val connectionProvider = getDSLContext.configuration().connectionProvider() - val connection = connectionProvider.acquire() + val connection = newRawConnection() connection.setAutoCommit(false) try { @@ -2244,7 +2242,7 @@ class DatasetResourceSpec assertStatus(ex, 409) } finally { connection.rollback() - connectionProvider.release(connection) + connection.close() } uploadPart(filePath, 1, minPartBytes(3.toByte)).getStatus shouldEqual 200 @@ -2255,8 +2253,7 @@ class DatasetResourceSpec initUpload(filePath, numParts = 2) val uploadId = fetchUploadIdOrFail(filePath) - val connectionProvider = getDSLContext.configuration().connectionProvider() - val connection = connectionProvider.acquire() + val connection = newRawConnection() connection.setAutoCommit(false) try { @@ -2274,7 +2271,7 @@ class DatasetResourceSpec uploadPart(filePath, 2, tinyBytes(9.toByte)).getStatus shouldEqual 200 } finally { connection.rollback() - connectionProvider.release(connection) + connection.close() } } @@ -2461,8 +2458,7 @@ class DatasetResourceSpec initUpload(filePath, numParts = 1) uploadPart(filePath, 1, tinyBytes(1.toByte)).getStatus shouldEqual 200 - val connectionProvider = getDSLContext.configuration().connectionProvider() - val connection = connectionProvider.acquire() + val connection = newRawConnection() connection.setAutoCommit(false) try { @@ -2482,7 +2478,7 @@ class DatasetResourceSpec assertStatus(ex, 409) } finally { connection.rollback() - connectionProvider.release(connection) + connection.close() } } @@ -2522,8 +2518,7 @@ class DatasetResourceSpec val filePath = uniqueFilePath("abort-lock-race") initUpload(filePath, numParts = 1) - val connectionProvider = getDSLContext.configuration().connectionProvider() - val connection = connectionProvider.acquire() + val connection = newRawConnection() connection.setAutoCommit(false) try { @@ -2543,7 +2538,7 @@ class DatasetResourceSpec assertStatus(ex, 409) } finally { connection.rollback() - connectionProvider.release(connection) + connection.close() } } diff --git a/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala b/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala index f54995b764..2445af02f7 100644 --- a/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala @@ -155,7 +155,7 @@ class StagedFileCleanupJobSpec } override protected def afterAll(): Unit = { - try shutdownDB() + try closeConnectionPool() finally super.afterAll() } diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala index a36ecdd590..35c3f47377 100644 --- a/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala +++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala @@ -79,7 +79,7 @@ class NotebookMigrationResourceSpec """{"operator_to_cell":{},"cell_to_operator":{}}""" override protected def beforeAll(): Unit = initializeDBAndReplaceDSLContext() - override protected def afterAll(): Unit = shutdownDB() + override protected def afterAll(): Unit = closeConnectionPool() override protected def beforeEach(): Unit = { val cfg = getDSLContext.configuration()
