Xiao-zhen-Liu commented on code in PR #6238:
URL: https://github.com/apache/texera/pull/6238#discussion_r3561625405
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
Review Comment:
If the `CREATE DATABASE` split marker ever changes, `parts.length > 1` is
false and this silently runs the whole DDL as the body. A hard failure would be
safer than quietly doing the wrong thing.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
+ val ds = new HikariDataSource(hikariConfig)
+ dataSource = Some(ds)
+
+ val jooqCfg = new DefaultConfiguration()
+ jooqCfg.set(new DataSourceConnectionProvider(ds))
+ jooqCfg.set(SQLDialect.POSTGRES)
+ val scopedCtx = DSL.using(jooqCfg)
+ testScopedContext = Some(scopedCtx)
+
+ // Point the Texera backend exactly to this suite's isolated database
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
+ SqlServer.getInstance().replaceDSLContext(scopedCtx)
}
- 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)
+ }
+
+ abstract override def withFixture(test: NoArgTest): Outcome = {
+ initializeDBAndReplaceDSLContext()
+
val sqlServerInstance = SqlServer.getInstance()
- dslContext = Some(DSL.using(texeraDB, SQLDialect.POSTGRES))
+ val activeContext = testScopedContext.get
- sqlServerInstance.replaceDSLContext(dslContext.get)
+ try {
+ sqlServerInstance.replaceDSLContext(activeContext)
+ super.withFixture(test)
+ } finally {
+ try {
+ // Truncate all tables on a pooled connection so each test starts
clean.
+ Using.resource(dataSource.get.getConnection) { conn =>
+ Using.resource(conn.createStatement()) { stmt =>
+ stmt.execute(
+ """
+ DO $$ DECLARE
+ r RECORD;
+ BEGIN
+ FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname =
'public') LOOP
Review Comment:
**This per-test cleanup is a silent no-op (targets the wrong schema).** It
truncates the `public` schema, but every table lives in `texera_db`
(`sql/texera_ddl.sql:42-43` creates that schema and sets `search_path`), so
`pg_tables WHERE schemaname='public'` matches nothing. Verified on JDK 11:
`ComputingUnitAccessSpec` and `FileResolverSpec` pass only because the
`beforeAll` seed is never deleted; retargeting the truncation at `texera_db`
makes `ComputingUnitAccessSpec` fail 3/5. Two consequences: (1) "each test
starts clean" is false today — the trait adds no per-test isolation, which is a
trap for anyone who later drops a suite's own cleanup trusting this; (2) you
can't just fix the schema name, because that breaks every suite that seeds only
in `beforeAll` (`AccessControlResourceSpec`, `ComputingUnitAccessSpec`,
`WorkflowResourceSpec`, `FileResolverSpec`, and the `DatasetResourceSpec` this
PR touches). Suggest either dropping the truncation and keeping each suite's
existing cl
eanup, or making it real and moving those seeds to `beforeEach`.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
+ val ds = new HikariDataSource(hikariConfig)
+ dataSource = Some(ds)
+
+ val jooqCfg = new DefaultConfiguration()
+ jooqCfg.set(new DataSourceConnectionProvider(ds))
+ jooqCfg.set(SQLDialect.POSTGRES)
+ val scopedCtx = DSL.using(jooqCfg)
+ testScopedContext = Some(scopedCtx)
+
+ // Point the Texera backend exactly to this suite's isolated database
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
+ SqlServer.getInstance().replaceDSLContext(scopedCtx)
}
- 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)
+ }
+
+ abstract override def withFixture(test: NoArgTest): Outcome = {
+ initializeDBAndReplaceDSLContext()
+
val sqlServerInstance = SqlServer.getInstance()
- dslContext = Some(DSL.using(texeraDB, SQLDialect.POSTGRES))
+ val activeContext = testScopedContext.get
- sqlServerInstance.replaceDSLContext(dslContext.get)
+ try {
+ sqlServerInstance.replaceDSLContext(activeContext)
+ super.withFixture(test)
+ } finally {
+ try {
+ // Truncate all tables on a pooled connection so each test starts
clean.
+ Using.resource(dataSource.get.getConnection) { conn =>
+ Using.resource(conn.createStatement()) { stmt =>
+ stmt.execute(
+ """
+ DO $$ DECLARE
+ r RECORD;
+ BEGIN
+ FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname =
'public') LOOP
+ EXECUTE 'TRUNCATE TABLE ' || quote_ident(r.tablename) ||
' CASCADE';
+ END LOOP;
+ END $$;
+ """
+ )
+ }
+ }
+ } catch {
+ case e: Exception => e.printStackTrace()
Review Comment:
If the truncation throws it's only printed and swallowed, so the next test
can start against dirty state. If the cleanup stays, a failure to clean should
fail loudly rather than print-and-continue.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
Review Comment:
This credits the singleton with avoiding `OverlappingFileLockException` "in
parallel," but tests still run sequentially (`Tags.limit` is in place), so the
actual current win is booting Postgres once. Worth aligning the comment with
today's behavior.
##########
amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala:
##########
@@ -199,10 +198,28 @@ object TestUtils {
* Note such test cases need to clean up the database at the end of running
each test case.
*/
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"DROP DATABASE IF EXISTS $dbName")
+ stmt.execute(s"CREATE DATABASE $dbName")
+ }
+ }
+
+ val targetDbConn = embedded.getDatabase("postgres", dbName).getConnection
Review Comment:
A few things in this method: (1) it duplicates the "create DB + run DDL"
sequence from the trait (`MockTexeraDB.scala:101-114`) and the copies have
already drifted — worth one shared helper on `object MockTexeraDB`. (2)
`targetDbConn` is closed manually after the `Using.resource`, so it leaks if
`execute` throws — wrap it in `Using.resource` too. (3) The `DROP DATABASE IF
EXISTS` on a freshly generated UUID name (line 208) is dead. (4) This quietly
moves the e2e specs off the external test Postgres onto the embedded one — a
real (good) change worth a line in the description; it depends on #4179. I ran
the e2e specs this way on JDK 11 (46 tests ×3, no `Connection refused`), so the
switch looks safe for sequential runs.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
+ val ds = new HikariDataSource(hikariConfig)
+ dataSource = Some(ds)
+
+ val jooqCfg = new DefaultConfiguration()
+ jooqCfg.set(new DataSourceConnectionProvider(ds))
+ jooqCfg.set(SQLDialect.POSTGRES)
+ val scopedCtx = DSL.using(jooqCfg)
+ testScopedContext = Some(scopedCtx)
+
+ // Point the Texera backend exactly to this suite's isolated database
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
+ SqlServer.getInstance().replaceDSLContext(scopedCtx)
}
- 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)
+ }
+
+ abstract override def withFixture(test: NoArgTest): Outcome = {
+ initializeDBAndReplaceDSLContext()
+
val sqlServerInstance = SqlServer.getInstance()
- dslContext = Some(DSL.using(texeraDB, SQLDialect.POSTGRES))
+ val activeContext = testScopedContext.get
- sqlServerInstance.replaceDSLContext(dslContext.get)
+ try {
+ sqlServerInstance.replaceDSLContext(activeContext)
+ super.withFixture(test)
+ } finally {
+ try {
+ // Truncate all tables on a pooled connection so each test starts
clean.
+ Using.resource(dataSource.get.getConnection) { conn =>
+ Using.resource(conn.createStatement()) { stmt =>
+ stmt.execute(
+ """
+ DO $$ DECLARE
+ r RECORD;
+ BEGIN
+ FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname =
'public') LOOP
+ EXECUTE 'TRUNCATE TABLE ' || quote_ident(r.tablename) ||
' CASCADE';
+ END LOOP;
+ END $$;
+ """
+ )
+ }
+ }
+ } catch {
+ case e: Exception => e.printStackTrace()
+ }
+ }
}
+
+ def getDSLContext: DSLContext =
+ synchronized {
+ if (testScopedContext.isEmpty) {
+ initializeDBAndReplaceDSLContext()
+ }
+ testScopedContext.get
+ }
+
+ def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
Review Comment:
Minor readability: `getDBInstance` exists on both the object and this trait,
and there are two init entry points (`ensureInitialized` vs
`initializeDBAndReplaceDSLContext`). A line of doc on each boundary would save
a maintainer some head-scratching.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
+ val ds = new HikariDataSource(hikariConfig)
+ dataSource = Some(ds)
+
+ val jooqCfg = new DefaultConfiguration()
+ jooqCfg.set(new DataSourceConnectionProvider(ds))
+ jooqCfg.set(SQLDialect.POSTGRES)
+ val scopedCtx = DSL.using(jooqCfg)
+ testScopedContext = Some(scopedCtx)
+
+ // Point the Texera backend exactly to this suite's isolated database
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
Review Comment:
`SqlServer.initConnection` constructs a `SqlServer` whose constructor
eagerly opens its own 10-connection Hikari pool (`SqlServer.scala:62`,
`minimumIdle=2`), and the next line swaps in the trait's separate pool. So each
trait suite opens two pools and the `SqlServer`-owned one is never used.
Consider a `SqlServer` method that repoints its existing pool, or reuse that
pool instead of building a new `DataSourceConnectionProvider`.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
Review Comment:
`tablesAndIndexCreation` is never reassigned — can be a `val`.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
+ val ds = new HikariDataSource(hikariConfig)
+ dataSource = Some(ds)
+
+ val jooqCfg = new DefaultConfiguration()
+ jooqCfg.set(new DataSourceConnectionProvider(ds))
+ jooqCfg.set(SQLDialect.POSTGRES)
+ val scopedCtx = DSL.using(jooqCfg)
+ testScopedContext = Some(scopedCtx)
+
+ // Point the Texera backend exactly to this suite's isolated database
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
+ SqlServer.getInstance().replaceDSLContext(scopedCtx)
}
- 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)
+ }
+
+ abstract override def withFixture(test: NoArgTest): Outcome = {
+ initializeDBAndReplaceDSLContext()
+
val sqlServerInstance = SqlServer.getInstance()
- dslContext = Some(DSL.using(texeraDB, SQLDialect.POSTGRES))
+ val activeContext = testScopedContext.get
- sqlServerInstance.replaceDSLContext(dslContext.get)
+ try {
+ sqlServerInstance.replaceDSLContext(activeContext)
Review Comment:
Per-suite isolation is real for `getDSLContext`, but production DAOs read
through the single global `SqlServer`, which this repoints per test. That's
safe only while `Tags.limit(Tags.Test,1)` keeps suites sequential (still in all
12 build.sbt). Once #4525 removes that, two suites in one JVM race on this
global — and `initConnection` closing the prior instance's pool can drop a pool
another suite is using. Worth noting this PR is groundwork; parallel-safety
needs a follow-up that makes the context per-suite/thread.
##########
file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala:
##########
@@ -1233,8 +1233,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 = MockTexeraDB.getDBInstance.getDatabase("postgres",
uniqueDbName).getConnection
Review Comment:
Opening a raw connection via `MockTexeraDB.getDBInstance.getDatabase(...)`
works (jOOQ qualifies the schema), but it couples the test to the trait's
internals — a small `newRawConnection()` helper on the trait would read
cleaner, and the same pattern repeats at 5 other sites in this file. Minor:
`getConnection` + `setAutoCommit(false)` run before the `try`, so an early
throw would leak the connection.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
+ val ds = new HikariDataSource(hikariConfig)
+ dataSource = Some(ds)
+
+ val jooqCfg = new DefaultConfiguration()
+ jooqCfg.set(new DataSourceConnectionProvider(ds))
+ jooqCfg.set(SQLDialect.POSTGRES)
+ val scopedCtx = DSL.using(jooqCfg)
+ testScopedContext = Some(scopedCtx)
+
+ // Point the Texera backend exactly to this suite's isolated database
+ SqlServer.initConnection(jdbcUrl, "postgres", "")
+ SqlServer.getInstance().replaceDSLContext(scopedCtx)
}
- 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)
+ }
+
+ abstract override def withFixture(test: NoArgTest): Outcome = {
+ initializeDBAndReplaceDSLContext()
+
val sqlServerInstance = SqlServer.getInstance()
- dslContext = Some(DSL.using(texeraDB, SQLDialect.POSTGRES))
+ val activeContext = testScopedContext.get
- sqlServerInstance.replaceDSLContext(dslContext.get)
+ try {
+ sqlServerInstance.replaceDSLContext(activeContext)
+ super.withFixture(test)
+ } finally {
+ try {
+ // Truncate all tables on a pooled connection so each test starts
clean.
+ Using.resource(dataSource.get.getConnection) { conn =>
+ Using.resource(conn.createStatement()) { stmt =>
+ stmt.execute(
+ """
+ DO $$ DECLARE
+ r RECORD;
+ BEGIN
+ FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname =
'public') LOOP
+ EXECUTE 'TRUNCATE TABLE ' || quote_ident(r.tablename) ||
' CASCADE';
+ END LOOP;
+ END $$;
+ """
+ )
+ }
+ }
+ } catch {
+ case e: Exception => e.printStackTrace()
+ }
+ }
}
+
+ def getDSLContext: DSLContext =
+ synchronized {
+ if (testScopedContext.isEmpty) {
+ initializeDBAndReplaceDSLContext()
+ }
+ testScopedContext.get
+ }
+
+ def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
+
+ def shutdownDB(): Unit =
Review Comment:
Two small things: the name no longer matches its behavior (it closes the
per-suite pool, not a DB), and the created databases are never dropped (fine
now, matters once suites run in parallel). Also #6063 asked to no-op
`shutdownDB`; closing the pool is better, so a one-line note that you deviated
on purpose would help.
##########
common/dao/src/test/scala/org/apache/texera/dao/MockTexeraDB.scala:
##########
@@ -19,146 +19,173 @@
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.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
-
-trait MockTexeraDB {
-
- private var dbInstance: Option[EmbeddedPostgres] = None
- private var dslContext: Option[DSLContext] = None
- private val database: String = "texera_db"
+import scala.util.Using
+
+/**
+ * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that
mix
+ * in this trait share one Postgres instance for the lifetime of the JVM,
which
+ * avoids the OverlappingFileLockException that occurs when each spec tries to
+ * extract the embedded Postgres binaries into the same directory in parallel.
+ */
+object MockTexeraDB {
private val username: String = "postgres"
private val password: String = ""
- def executeScriptInJDBC(conn: Connection, script: String): Unit = {
- assert(dbInstance.nonEmpty)
- conn.prepareStatement(script).execute()
- conn.close()
- }
+ @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+ @volatile private var ddlScript: Option[String] = None
- def getDSLContext: DSLContext = {
- dslContext match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ def ensureInitialized(): Unit =
+ synchronized {
+ if (dbInstance.isDefined && ddlScript.isDefined) return
- def getDBInstance: EmbeddedPostgres = {
- dbInstance match {
- case Some(value) => value
- case None =>
- throw new RuntimeException(
- "test database is not initialized. Did you call
initializeDBAndReplaceDSLContext()?"
- )
- }
- }
+ if (dbInstance.isEmpty) {
+ val driver = new org.postgresql.Driver()
+ DriverManager.registerDriver(driver)
- def shutdownDB(): Unit = {
- dbInstance match {
- case Some(value) =>
- value.close()
- dbInstance = None
- dslContext = None
- case None =>
- // do nothing
- }
- }
+ // Boot the heavy JVM engine exactly once
+ dbInstance = Some(EmbeddedPostgres.builder().start())
+ }
+
+ val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+ val source = Source.fromFile(ddlPath.toString)
+ val content =
+ try source.mkString
+ finally source.close()
- def initializeDBAndReplaceDSLContext(): Unit = {
- assert(dbInstance.isEmpty && dslContext.isEmpty)
+ val parts: Array[String] = content.split("(?m)^CREATE DATABASE
:\"DB_NAME\";")
+ val sqlBody = if (parts.length > 1) parts(1) else content
- val driver = new org.postgresql.Driver()
- DriverManager.registerDriver(driver)
+ def removeCCommands(sql: String): String =
+ sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
- val embedded = EmbeddedPostgres.builder().start()
+ var tablesAndIndexCreation = removeCCommands(sqlBody)
- dbInstance = Some(embedded)
+ 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
- val ddlPath = {
- Paths.get("sql/texera_ddl.sql").toRealPath()
+ // Cache the cleaned script so parallel suites don't have to re-read the
file
+ ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation,
replacementText).trim)
}
- val source = Source.fromFile(ddlPath.toString)
- val content =
- try {
- source.mkString
- } finally {
- source.close()
+
+ def getDBInstance: EmbeddedPostgres =
+ dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+ def getDDLScript: String = ddlScript.getOrElse(throw new
RuntimeException("DDL not loaded"))
+}
+
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
+ private var testScopedContext: Option[DSLContext] = None
+ protected var dataSource: Option[HikariDataSource] = None
+ protected var uniqueDbName: String = ""
+
+ 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)
+
+ // Back the test DSLContext with a real HikariCP pool so that
concurrent
+ // transactions acquire *distinct* connections (matching production),
rather
+ // than trampling one shared connection's autoCommit flag.
+ val hikariConfig = new HikariConfig()
+ hikariConfig.setJdbcUrl(jdbcUrl)
+ hikariConfig.setUsername("postgres")
+ hikariConfig.setPassword("")
+ // Must exceed the maximum number of concurrent test threads.
+ hikariConfig.setMaximumPoolSize(10)
Review Comment:
`10` is justified by "must exceed the max concurrent test threads," but
nothing enforces that and exceeding it blocks 30s then fails. A named constant
with the rationale (or deriving it) would age better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
