From 3998b080fec12c58060690bb6b2e80cbbc474533 Mon Sep 17 00:00:00 2001
From: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>
Date: Sat, 25 Apr 2026 22:54:33 -0700
Subject: [PATCH 01/13] perf(test): make MockTexeraDB a JVM singleton

Each spec that mixes in MockTexeraDB used to start its own
EmbeddedPostgres in beforeAll. Once the global Tags.limit(Tags.Test, 1)
restriction was lifted, these specs ran in parallel and collided on
the shared /tmp/embedded-pg/<binary>/ directory with an
OverlappingFileLockException at EmbeddedPostgres.prepareBinaries.

Move the EmbeddedPostgres + DSLContext fields onto a companion object,
make ensureInitialized() idempotent and synchronized, and turn
shutdownDB() into a no-op so the singleton lives for the JVM. The
trait still exposes the same methods so spec callers do not change.
---
 .../org/apache/texera/dao/MockTexeraDB.scala  | 239 +++++++++---------
 1 file changed, 122 insertions(+), 117 deletions(-)

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 3ae97b62e03..ced7273538c 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
@@ -27,7 +27,13 @@ import java.nio.file.Paths
 import java.sql.{Connection, DriverManager}
 import scala.io.Source
 
-trait MockTexeraDB {
+/**
+  * 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 var dbInstance: Option[EmbeddedPostgres] = None
   private var dslContext: Option[DSLContext] = None
@@ -35,130 +41,129 @@ trait 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()
-  }
+  private var dbInstance: Option[EmbeddedPostgres] = None
+  private var dslContext: Option[DSLContext] = 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) return
+
+      val driver = new org.postgresql.Driver()
+      DriverManager.registerDriver(driver)
+
+      val embedded = EmbeddedPostgres.builder().start()
+      dbInstance = Some(embedded)
+
+      val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+      val source = Source.fromFile(ddlPath.toString)
+      val content =
+        try source.mkString
+        finally source.close()
+      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
+      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()
+      val ctx = DSL.using(texeraDB, SQLDialect.POSTGRES)
+      dslContext = Some(ctx)
+      sqlServerInstance.replaceDSLContext(ctx)
     }
-  }
 
-  def getDBInstance: EmbeddedPostgres = {
-    dbInstance match {
-      case Some(value) => value
-      case None =>
-        throw new RuntimeException(
-          "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?"
-        )
-    }
+  def getDSLContext: DSLContext =
+    dslContext.getOrElse(
+      throw new RuntimeException(
+        "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?"
+      )
+    )
+
+  def getDBInstance: EmbeddedPostgres =
+    dbInstance.getOrElse(
+      throw new RuntimeException(
+        "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?"
+      )
+    )
+
+  private def executeScriptInJDBC(conn: Connection, script: String): Unit = {
+    conn.prepareStatement(script).execute()
+    conn.close()
   }
+}
 
-  def shutdownDB(): Unit = {
-    dbInstance match {
-      case Some(value) =>
-        value.close()
-        dbInstance = None
-        dslContext = None
-      case None =>
-      // do nothing
-    }
-  }
+trait MockTexeraDB {
 
-  def initializeDBAndReplaceDSLContext(): Unit = {
-    assert(dbInstance.isEmpty && dslContext.isEmpty)
+  def executeScriptInJDBC(conn: Connection, script: String): Unit = {
+    conn.prepareStatement(script).execute()
+    conn.close()
+  }
 
-    val driver = new org.postgresql.Driver()
-    DriverManager.registerDriver(driver)
+  def getDSLContext: DSLContext = MockTexeraDB.getDSLContext
 
-    val embedded = EmbeddedPostgres.builder().start()
+  def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
 
-    dbInstance = Some(embedded)
+  def initializeDBAndReplaceDSLContext(): Unit = MockTexeraDB.ensureInitialized()
 
-    val ddlPath = {
-      Paths.get("sql/texera_ddl.sql").toRealPath()
-    }
-    val source = Source.fromFile(ddlPath.toString)
-    val content =
-      try {
-        source.mkString
-      } finally {
-        source.close()
-      }
-    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))
-
-    sqlServerInstance.replaceDSLContext(dslContext.get)
-  }
+  /**
+    * No-op. The singleton EmbeddedPostgres lives for the lifetime of the JVM,
+    * so individual specs should not shut it down. Kept for API compatibility
+    * with existing afterAll hooks.
+    */
+  def shutdownDB(): Unit = ()
 }

From 556ecd75df7d2458507ae1aa9b35274eed973878 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Thu, 2 Jul 2026 11:24:24 -0700
Subject: [PATCH 02/13] MockTexeraDB cherrypick

---
 .../src/test/scala/org/apache/texera/dao/MockTexeraDB.scala    | 3 ---
 1 file changed, 3 deletions(-)

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 ced7273538c..a84d537d8d2 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
@@ -34,9 +34,6 @@ import scala.io.Source
   * extract the embedded Postgres binaries into the same directory in parallel.
   */
 object MockTexeraDB {
-
-  private var dbInstance: Option[EmbeddedPostgres] = None
-  private var dslContext: Option[DSLContext] = None
   private val database: String = "texera_db"
   private val username: String = "postgres"
   private val password: String = ""

From f7bc8831af456832d99176def91460bf63c8ccc6 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Mon, 6 Jul 2026 12:36:46 -0700
Subject: [PATCH 03/13] perf (test): Add transactional rollbacks to
 MockTexeraDB

---
 .../org/apache/texera/dao/MockTexeraDB.scala  | 43 +++++++++++++++++--
 1 file changed, 40 insertions(+), 3 deletions(-)

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 a84d537d8d2..a1e4627e2f5 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
@@ -22,6 +22,8 @@ package org.apache.texera.dao
 import io.zonky.test.db.postgres.embedded.EmbeddedPostgres
 import org.jooq.impl.DSL
 import org.jooq.{DSLContext, SQLDialect}
+import org.scalatest.{Outcome, TestSuiteMixin}
+import org.scalatest.flatspec.{AnyFlatSpec, AnyFlatSpecLike}
 
 import java.nio.file.Paths
 import java.sql.{Connection, DriverManager}
@@ -144,18 +146,53 @@ object MockTexeraDB {
   }
 }
 
-trait MockTexeraDB {
+trait MockTexeraDB extends AnyFlatSpecLike {
+  private var testScopedContext : Option[DSLContext] = None
+  protected var connection : Option[Connection]
+
+
+  override def withFixture(test: NoArgTest) : Outcome = {
+    initializeDBAndReplaceDSLContext()
+    val sqlServerInstance = SqlServer.getInstance()
+    val baseContext = MockTexeraDB.getDSLContext
+
+    try{
+      sqlServerInstance.replaceDSLContext(testScopedContext.get)
+      super.withFixture(test)
+
+    }finally {
+      try {
+        val conn = connection.get
+        if (!conn.isClosed) {
+          conn.rollback()
+          conn.close()
+        }
+      } catch {
+        case e: Exception => e.printStackTrace()
+      }
+      sqlServerInstance.replaceDSLContext(baseContext)
+    }
+  }
 
   def executeScriptInJDBC(conn: Connection, script: String): Unit = {
     conn.prepareStatement(script).execute()
     conn.close()
   }
 
-  def getDSLContext: DSLContext = MockTexeraDB.getDSLContext
+  def getDSLContext: DSLContext =
+    testScopedContext.getOrElse(MockTexeraDB.getDSLContext)
 
   def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
 
-  def initializeDBAndReplaceDSLContext(): Unit = MockTexeraDB.ensureInitialized()
+  def initializeDBAndReplaceDSLContext(): Unit = {
+    val embedded = MockTexeraDB.getDBInstance
+    connection = Some(embedded.getDatabase("postgres", "texera_db").getConnection)
+    connection.get.setAutoCommit(false)
+
+    testScopedContext = Some(DSL.using(connection.get, SQLDialect.POSTGRES))
+
+    MockTexeraDB.ensureInitialized()
+  }
 
   /**
     * No-op. The singleton EmbeddedPostgres lives for the lifetime of the JVM,

From cc351fae1eddb20f7b390b18fa27ff5cac933e01 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Mon, 6 Jul 2026 12:56:33 -0700
Subject: [PATCH 04/13] perf (test): MockTexeraDB bug fixes

---
 .../test/scala/org/apache/texera/dao/MockTexeraDB.scala   | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

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 a1e4627e2f5..3c3c7122cd0 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
@@ -148,8 +148,7 @@ object MockTexeraDB {
 
 trait MockTexeraDB extends AnyFlatSpecLike {
   private var testScopedContext : Option[DSLContext] = None
-  protected var connection : Option[Connection]
-
+  private var connection : Option[Connection] = None
 
   override def withFixture(test: NoArgTest) : Outcome = {
     initializeDBAndReplaceDSLContext()
@@ -185,13 +184,12 @@ trait MockTexeraDB extends AnyFlatSpecLike {
   def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
 
   def initializeDBAndReplaceDSLContext(): Unit = {
+    MockTexeraDB.ensureInitialized()
+
     val embedded = MockTexeraDB.getDBInstance
     connection = Some(embedded.getDatabase("postgres", "texera_db").getConnection)
     connection.get.setAutoCommit(false)
-
     testScopedContext = Some(DSL.using(connection.get, SQLDialect.POSTGRES))
-
-    MockTexeraDB.ensureInitialized()
   }
 
   /**

From 6cf34e0169be3834c7b7548abab5c60fa55210f1 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Mon, 6 Jul 2026 13:56:19 -0700
Subject: [PATCH 05/13] perf (test): Fake databases

---
 .../org/apache/texera/dao/MockTexeraDB.scala  | 248 ++++++++----------
 1 file changed, 110 insertions(+), 138 deletions(-)

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 3c3c7122cd0..76c84777987 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
@@ -22,12 +22,13 @@ package org.apache.texera.dao
 import io.zonky.test.db.postgres.embedded.EmbeddedPostgres
 import org.jooq.impl.DSL
 import org.jooq.{DSLContext, SQLDialect}
-import org.scalatest.{Outcome, TestSuiteMixin}
-import org.scalatest.flatspec.{AnyFlatSpec, AnyFlatSpecLike}
+import org.scalatest.Outcome
+import org.scalatest.flatspec.AnyFlatSpecLike
 
 import java.nio.file.Paths
-import java.sql.{Connection, DriverManager}
+import java.sql.{Connection, DriverManager, Savepoint}
 import scala.io.Source
+import scala.util.Using
 
 /**
   * Provides a JVM-singleton EmbeddedPostgres for tests. Multiple specs that mix
@@ -36,166 +37,137 @@ import scala.io.Source
   * extract the embedded Postgres binaries into the same directory in parallel.
   */
 object MockTexeraDB {
-  private val database: String = "texera_db"
   private val username: String = "postgres"
   private val password: String = ""
 
-  private var dbInstance: Option[EmbeddedPostgres] = None
-  private var dslContext: Option[DSLContext] = None
-
-  def ensureInitialized(): Unit =
-    synchronized {
-      if (dbInstance.isDefined) return
-
-      val driver = new org.postgresql.Driver()
-      DriverManager.registerDriver(driver)
-
-      val embedded = EmbeddedPostgres.builder().start()
-      dbInstance = Some(embedded)
-
-      val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
-      val source = Source.fromFile(ddlPath.toString)
-      val content =
-        try source.mkString
-        finally source.close()
-      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
-      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()
-      val ctx = DSL.using(texeraDB, SQLDialect.POSTGRES)
-      dslContext = Some(ctx)
-      sqlServerInstance.replaceDSLContext(ctx)
-    }
+  @volatile private var dbInstance: Option[EmbeddedPostgres] = None
+  @volatile private var ddlScript: Option[String] = None
+
+  def ensureInitialized(): Unit = synchronized {
+    if (dbInstance.isDefined) return
+
+    val driver = new org.postgresql.Driver()
+    DriverManager.registerDriver(driver)
+
+    // Boot the heavy JVM engine exactly once
+    val embedded = EmbeddedPostgres.builder().start()
+    dbInstance = Some(embedded)
+
+    val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+    val source = Source.fromFile(ddlPath.toString)
+    val content = try source.mkString finally source.close()
+
+    val parts: Array[String] = content.split("(?m)^CREATE DATABASE :\"DB_NAME\";")
+    val sqlBody = if (parts.length > 1) parts(1) else content
+
+    def removeCCommands(sql: String): String =
+      sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
+
+    var 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
 
-  def getDSLContext: DSLContext =
-    dslContext.getOrElse(
-      throw new RuntimeException(
-        "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?"
-      )
-    )
-
-  def getDBInstance: EmbeddedPostgres =
-    dbInstance.getOrElse(
-      throw new RuntimeException(
-        "test database is not initialized. Did you call initializeDBAndReplaceDSLContext()?"
-      )
-    )
-
-  private def executeScriptInJDBC(conn: Connection, script: String): Unit = {
-    conn.prepareStatement(script).execute()
-    conn.close()
+    // Cache the cleaned script so parallel suites don't have to re-read the file
+    ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation, replacementText).trim)
   }
+
+  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 AnyFlatSpecLike {
-  private var testScopedContext : Option[DSLContext] = None
-  private var connection : Option[Connection] = None
+  private var testScopedContext: Option[DSLContext] = None
+  private var connection: Option[Connection] = None
+  private var uniqueDbName: String = ""
+
+  def initializeDBAndReplaceDSLContext(): Unit = synchronized {
+    if (connection.isEmpty || connection.get.isClosed) {
+      MockTexeraDB.ensureInitialized()
+      val embedded = MockTexeraDB.getDBInstance
+
+      // 1. Create a 100% isolated logical database for this specific test suite
+      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")
+        }
+      }
+
+      // 2. Connect to the isolated DB and apply the DDL
+      val conn = embedded.getDatabase("postgres", uniqueDbName).getConnection
 
-  override def withFixture(test: NoArgTest) : Outcome = {
+      // AutoCommit is TRUE by default, meaning any records inserted in beforeAll()
+      // will be permanently committed to this suite's specific isolated database!
+      Using.resource(conn.createStatement()) { stmt =>
+        stmt.execute(MockTexeraDB.getDDLScript)
+      }
+
+      connection = Some(conn)
+      val scopedCtx = DSL.using(conn, SQLDialect.POSTGRES)
+      testScopedContext = Some(scopedCtx)
+
+      // Point the Texera backend exactly to this suite's isolated database
+      SqlServer.initConnection(embedded.getJdbcUrl("postgres", uniqueDbName), "postgres", "")
+      SqlServer.getInstance().replaceDSLContext(scopedCtx)
+    }
+  }
+
+  override def withFixture(test: NoArgTest): Outcome = {
     initializeDBAndReplaceDSLContext()
+
+    val conn = connection.get
     val sqlServerInstance = SqlServer.getInstance()
-    val baseContext = MockTexeraDB.getDSLContext
+    val activeContext = testScopedContext.get
 
-    try{
-      sqlServerInstance.replaceDSLContext(testScopedContext.get)
-      super.withFixture(test)
+    // 3. Turn OFF autocommit to sandbox this individual test case
+    conn.setAutoCommit(false)
 
-    }finally {
+    try {
+      sqlServerInstance.replaceDSLContext(activeContext)
+      super.withFixture(test)
+    } finally {
       try {
-        val conn = connection.get
         if (!conn.isClosed) {
+          // 4. Perform a FULL rollback. This wipes dirty test data and clears any
+          // aborted SQL errors. Because beforeAll data was already auto-committed, it survives!
           conn.rollback()
-          conn.close()
+          // 5. Restore autoCommit for afterEach / afterAll hooks
+          conn.setAutoCommit(true)
         }
       } catch {
         case e: Exception => e.printStackTrace()
       }
-      sqlServerInstance.replaceDSLContext(baseContext)
     }
   }
 
-  def executeScriptInJDBC(conn: Connection, script: String): Unit = {
-    conn.prepareStatement(script).execute()
-    conn.close()
+  def getDSLContext: DSLContext = synchronized {
+    if (testScopedContext.isEmpty) {
+      initializeDBAndReplaceDSLContext()
+    }
+    testScopedContext.get
   }
 
-  def getDSLContext: DSLContext =
-    testScopedContext.getOrElse(MockTexeraDB.getDSLContext)
-
   def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
 
-  def initializeDBAndReplaceDSLContext(): Unit = {
-    MockTexeraDB.ensureInitialized()
-
-    val embedded = MockTexeraDB.getDBInstance
-    connection = Some(embedded.getDatabase("postgres", "texera_db").getConnection)
-    connection.get.setAutoCommit(false)
-    testScopedContext = Some(DSL.using(connection.get, SQLDialect.POSTGRES))
+  def shutdownDB(): Unit = synchronized {
+    try {
+      connection.foreach { conn =>
+        if (!conn.isClosed) {
+          conn.close()
+        }
+      }
+    } catch {
+      case e: Exception => e.printStackTrace()
+    } finally {
+      connection = None
+      testScopedContext = None
+    }
   }
-
-  /**
-    * No-op. The singleton EmbeddedPostgres lives for the lifetime of the JVM,
-    * so individual specs should not shut it down. Kept for API compatibility
-    * with existing afterAll hooks.
-    */
-  def shutdownDB(): Unit = ()
-}
+}
\ No newline at end of file

From 9738b11452fff1cae6f4744a15d14a587941faa9 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Mon, 6 Jul 2026 14:15:28 -0700
Subject: [PATCH 06/13] perf (test): Fake databases

---
 .../org/apache/texera/dao/MockTexeraDB.scala  | 21 ++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

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 76c84777987..a9c6a0cabde 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
@@ -126,8 +126,7 @@ trait MockTexeraDB extends AnyFlatSpecLike {
     val sqlServerInstance = SqlServer.getInstance()
     val activeContext = testScopedContext.get
 
-    // 3. Turn OFF autocommit to sandbox this individual test case
-    conn.setAutoCommit(false)
+    conn.setAutoCommit(true)
 
     try {
       sqlServerInstance.replaceDSLContext(activeContext)
@@ -135,11 +134,19 @@ trait MockTexeraDB extends AnyFlatSpecLike {
     } finally {
       try {
         if (!conn.isClosed) {
-          // 4. Perform a FULL rollback. This wipes dirty test data and clears any
-          // aborted SQL errors. Because beforeAll data was already auto-committed, it survives!
-          conn.rollback()
-          // 5. Restore autoCommit for afterEach / afterAll hooks
-          conn.setAutoCommit(true)
+          scala.util.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()

From 30d8af2abf0db3fd91ea6b4e90350b4ee24e753e Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Mon, 6 Jul 2026 14:47:21 -0700
Subject: [PATCH 07/13] perf (test): Bug fixes

---
 .../texera/amber/engine/e2e/TestUtils.scala   | 24 ++++++++++++--
 .../org/apache/texera/dao/MockTexeraDB.scala  |  2 --
 .../resource/DatasetResourceSpec.scala        | 31 ++++++++-----------
 3 files changed, 34 insertions(+), 23 deletions(-)

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 ac71483a5d3..b1cc0aa6126 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
@@ -199,10 +199,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"
+
+    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
+    scala.util.Using.resource(targetDbConn.createStatement()) { stmt =>
+      stmt.execute(org.apache.texera.dao.MockTexeraDB.getDDLScript)
+    }
+    targetDbConn.close()
+
     SqlServer.initConnection(
-      StorageConfig.jdbcUrlForTestCases,
-      StorageConfig.jdbcUsername,
-      StorageConfig.jdbcPassword
+      embedded.getJdbcUrl("postgres", dbName),
+      "postgres",
+      ""
     )
   }
 
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 a9c6a0cabde..609c2736bc8 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
@@ -92,7 +92,6 @@ trait MockTexeraDB extends AnyFlatSpecLike {
       MockTexeraDB.ensureInitialized()
       val embedded = MockTexeraDB.getDBInstance
 
-      // 1. Create a 100% isolated logical database for this specific test suite
       uniqueDbName = "texera_db_" + java.util.UUID.randomUUID().toString.replace("-", "")
       Using.resource(embedded.getPostgresDatabase.getConnection) { defaultConn =>
         Using.resource(defaultConn.createStatement()) { stmt =>
@@ -100,7 +99,6 @@ trait MockTexeraDB extends AnyFlatSpecLike {
         }
       }
 
-      // 2. Connect to the isolated DB and apply the DDL
       val conn = embedded.getDatabase("postgres", uniqueDbName).getConnection
 
       // AutoCommit is TRUE by default, meaning any records inserted in beforeAll()
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 1730d12a0a0..174b80d2707 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
@@ -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 = getDBInstance.getPostgresDatabase.getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1256,7 +1255,7 @@ class DatasetResourceSpec
       ex.getResponse.getStatus shouldEqual 409
     } finally {
       connection.rollback()
-      connectionProvider.release(connection)
+      connection.close()
     }
 
     // lock released => init works again
@@ -1622,8 +1621,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 = getDBInstance.getPostgresDatabase.getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1645,7 +1644,7 @@ class DatasetResourceSpec
       ex.getResponse.getStatus shouldEqual 409
     } finally {
       connection.rollback()
-      connectionProvider.release(connection)
+      connection.close()
     }
   }
 
@@ -1865,8 +1864,7 @@ class DatasetResourceSpec
     initUpload(filePath, numParts = 2)
     val uploadId = fetchUploadIdOrFail(filePath)
 
-    val connectionProvider = getDSLContext.configuration().connectionProvider()
-    val connection = connectionProvider.acquire()
+    val connection = getDBInstance.getPostgresDatabase.getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1887,7 +1885,7 @@ class DatasetResourceSpec
       assertStatus(ex, 409)
     } finally {
       connection.rollback()
-      connectionProvider.release(connection)
+      connection.close()
     }
 
     uploadPart(filePath, 1, minPartBytes(3.toByte)).getStatus shouldEqual 200
@@ -1898,8 +1896,7 @@ class DatasetResourceSpec
     initUpload(filePath, numParts = 2)
     val uploadId = fetchUploadIdOrFail(filePath)
 
-    val connectionProvider = getDSLContext.configuration().connectionProvider()
-    val connection = connectionProvider.acquire()
+    val connection = getDBInstance.getPostgresDatabase.getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1917,7 +1914,7 @@ class DatasetResourceSpec
       uploadPart(filePath, 2, tinyBytes(9.toByte)).getStatus shouldEqual 200
     } finally {
       connection.rollback()
-      connectionProvider.release(connection)
+      connection.close()
     }
   }
 
@@ -2104,8 +2101,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 = getDBInstance.getPostgresDatabase.getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -2125,7 +2121,7 @@ class DatasetResourceSpec
       assertStatus(ex, 409)
     } finally {
       connection.rollback()
-      connectionProvider.release(connection)
+      connection.close()
     }
   }
 
@@ -2165,8 +2161,7 @@ class DatasetResourceSpec
     val filePath = uniqueFilePath("abort-lock-race")
     initUpload(filePath, numParts = 1)
 
-    val connectionProvider = getDSLContext.configuration().connectionProvider()
-    val connection = connectionProvider.acquire()
+    val connection = getDBInstance.getPostgresDatabase.getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -2186,7 +2181,7 @@ class DatasetResourceSpec
       assertStatus(ex, 409)
     } finally {
       connection.rollback()
-      connectionProvider.release(connection)
+      connection.close()
     }
   }
 

From 03aa1fdddc0470799c8a4593034cb28562f37921 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Tue, 7 Jul 2026 13:59:22 -0700
Subject: [PATCH 08/13] fix (build): fix `withTransaction` to handle disabling
 autoCommits.

---
 .../texera/amber/engine/e2e/TestUtils.scala   |  2 +-
 .../org/apache/texera/dao/SqlServer.scala     | 36 +++++++++++++++----
 .../org/apache/texera/dao/MockTexeraDB.scala  |  6 ++--
 .../resource/DatasetResourceSpec.scala        | 12 +++----
 4 files changed, 40 insertions(+), 16 deletions(-)

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 b1cc0aa6126..b71ac0e749c 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
@@ -202,7 +202,7 @@ object TestUtils {
     org.apache.texera.dao.MockTexeraDB.ensureInitialized()
     val embedded = org.apache.texera.dao.MockTexeraDB.getDBInstance
 
-    val dbName = "texera_db_for_test_cases"
+    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 =>
diff --git a/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala b/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
index 6348ae41fa0..9192804efc1 100644
--- a/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
+++ b/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
@@ -94,13 +94,37 @@ object SqlServer {
     * @return
     */
   def withTransaction[T](dsl: DSLContext)(block: DSLContext => T): T = {
-    var result: Option[T] = None
+    val provider = dsl.configuration().connectionProvider()
+    val conn = provider.acquire()
+    val originalAutoCommit = conn.getAutoCommit
 
-    dsl.transaction(configuration => {
-      val ctx = DSL.using(configuration)
-      result = Some(block(ctx))
-    })
+    try {
+      if (originalAutoCommit) {
+        conn.setAutoCommit(false)
+      }
 
-    result.getOrElse(throw new RuntimeException("Transaction failed without result!"))
+      val txCtx = org.jooq.impl.DSL.using(conn, dsl.dialect())
+      val result = block(txCtx)
+
+      if (originalAutoCommit) {
+        conn.commit()
+      }
+
+      result
+    } catch {
+      case e: Throwable =>
+        if (originalAutoCommit) {
+          conn.rollback()
+        }
+        throw e
+    } finally {
+      try {
+        if (originalAutoCommit != conn.getAutoCommit) {
+          conn.setAutoCommit(originalAutoCommit)
+        }
+      } finally {
+        provider.release(conn)
+      }
+    }
   }
 }
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 609c2736bc8..57d2f28421f 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
@@ -26,7 +26,7 @@ import org.scalatest.Outcome
 import org.scalatest.flatspec.AnyFlatSpecLike
 
 import java.nio.file.Paths
-import java.sql.{Connection, DriverManager, Savepoint}
+import java.sql.{Connection, DriverManager}
 import scala.io.Source
 import scala.util.Using
 
@@ -84,8 +84,8 @@ object MockTexeraDB {
 
 trait MockTexeraDB extends AnyFlatSpecLike {
   private var testScopedContext: Option[DSLContext] = None
-  private var connection: Option[Connection] = None
-  private var uniqueDbName: String = ""
+  protected var connection: Option[Connection] = None
+  protected var uniqueDbName: String = ""
 
   def initializeDBAndReplaceDSLContext(): Unit = synchronized {
     if (connection.isEmpty || connection.get.isClosed) {
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 174b80d2707..50088abc161 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
@@ -1233,7 +1233,7 @@ class DatasetResourceSpec
     val filePath = uniqueFilePath("init-session-row-locked")
     initUpload(filePath, numParts = 2).getStatus shouldEqual 200
 
-    val connection = getDBInstance.getPostgresDatabase.getConnection
+    val connection = MockTexeraDB.getDBInstance.getDatabase("postgres", uniqueDbName).getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1622,7 +1622,7 @@ class DatasetResourceSpec
     initUpload(filePath, numParts = 2).getStatus shouldEqual 200
 
     // Open a completely independent connection to simulate a second concurrent user
-    val connection = getDBInstance.getPostgresDatabase.getConnection
+    val connection = MockTexeraDB.getDBInstance.getDatabase("postgres", uniqueDbName).getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1864,7 +1864,7 @@ class DatasetResourceSpec
     initUpload(filePath, numParts = 2)
     val uploadId = fetchUploadIdOrFail(filePath)
 
-    val connection = getDBInstance.getPostgresDatabase.getConnection
+    val connection = MockTexeraDB.getDBInstance.getDatabase("postgres", uniqueDbName).getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -1896,7 +1896,7 @@ class DatasetResourceSpec
     initUpload(filePath, numParts = 2)
     val uploadId = fetchUploadIdOrFail(filePath)
 
-    val connection = getDBInstance.getPostgresDatabase.getConnection
+    val connection = MockTexeraDB.getDBInstance.getDatabase("postgres", uniqueDbName).getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -2101,7 +2101,7 @@ class DatasetResourceSpec
     initUpload(filePath, numParts = 1)
     uploadPart(filePath, 1, tinyBytes(1.toByte)).getStatus shouldEqual 200
 
-    val connection = getDBInstance.getPostgresDatabase.getConnection
+    val connection = MockTexeraDB.getDBInstance.getDatabase("postgres", uniqueDbName).getConnection
     connection.setAutoCommit(false)
 
     try {
@@ -2161,7 +2161,7 @@ class DatasetResourceSpec
     val filePath = uniqueFilePath("abort-lock-race")
     initUpload(filePath, numParts = 1)
 
-    val connection = getDBInstance.getPostgresDatabase.getConnection
+    val connection = MockTexeraDB.getDBInstance.getDatabase("postgres", uniqueDbName).getConnection
     connection.setAutoCommit(false)
 
     try {

From 46ef429b27840d4b08d56b04180e558011ae916e Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Tue, 7 Jul 2026 14:09:46 -0700
Subject: [PATCH 09/13] style(backend): run scalafmt to format Scala files

---
 .../org/apache/texera/dao/MockTexeraDB.scala  | 131 +++++++++---------
 1 file changed, 69 insertions(+), 62 deletions(-)

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 57d2f28421f..dce303611e8 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
@@ -43,42 +43,46 @@ object MockTexeraDB {
   @volatile private var dbInstance: Option[EmbeddedPostgres] = None
   @volatile private var ddlScript: Option[String] = None
 
-  def ensureInitialized(): Unit = synchronized {
-    if (dbInstance.isDefined) return
+  def ensureInitialized(): Unit =
+    synchronized {
+      if (dbInstance.isDefined) return
 
-    val driver = new org.postgresql.Driver()
-    DriverManager.registerDriver(driver)
+      val driver = new org.postgresql.Driver()
+      DriverManager.registerDriver(driver)
 
-    // Boot the heavy JVM engine exactly once
-    val embedded = EmbeddedPostgres.builder().start()
-    dbInstance = Some(embedded)
+      // Boot the heavy JVM engine exactly once
+      val embedded = EmbeddedPostgres.builder().start()
+      dbInstance = Some(embedded)
 
-    val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
-    val source = Source.fromFile(ddlPath.toString)
-    val content = try source.mkString finally source.close()
+      val ddlPath = Paths.get("sql/texera_ddl.sql").toRealPath()
+      val source = Source.fromFile(ddlPath.toString)
+      val content =
+        try source.mkString
+        finally source.close()
 
-    val parts: Array[String] = content.split("(?m)^CREATE DATABASE :\"DB_NAME\";")
-    val sqlBody = if (parts.length > 1) parts(1) else content
+      val parts: Array[String] = content.split("(?m)^CREATE DATABASE :\"DB_NAME\";")
+      val sqlBody = if (parts.length > 1) parts(1) else content
 
-    def removeCCommands(sql: String): String =
-      sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
+      def removeCCommands(sql: String): String =
+        sql.linesIterator.filterNot(_.trim.startsWith("\\c")).mkString("\n")
 
-    var tablesAndIndexCreation = removeCCommands(sqlBody)
+      var 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, '')));
+      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)
-  }
+      // Cache the cleaned script so parallel suites don't have to re-read the file
+      ddlScript = Some(blockPattern.replaceAllIn(tablesAndIndexCreation, replacementText).trim)
+    }
 
-  def getDBInstance: EmbeddedPostgres = dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
+  def getDBInstance: EmbeddedPostgres =
+    dbInstance.getOrElse(throw new RuntimeException("DB not initialized"))
   def getDDLScript: String = ddlScript.getOrElse(throw new RuntimeException("DDL not loaded"))
 }
 
@@ -87,35 +91,36 @@ trait MockTexeraDB extends AnyFlatSpecLike {
   protected var connection: Option[Connection] = None
   protected var uniqueDbName: String = ""
 
-  def initializeDBAndReplaceDSLContext(): Unit = synchronized {
-    if (connection.isEmpty || connection.get.isClosed) {
-      MockTexeraDB.ensureInitialized()
-      val embedded = MockTexeraDB.getDBInstance
+  def initializeDBAndReplaceDSLContext(): Unit =
+    synchronized {
+      if (connection.isEmpty || connection.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")
+        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")
+          }
         }
-      }
 
-      val conn = embedded.getDatabase("postgres", uniqueDbName).getConnection
+        val conn = embedded.getDatabase("postgres", uniqueDbName).getConnection
 
-      // AutoCommit is TRUE by default, meaning any records inserted in beforeAll()
-      // will be permanently committed to this suite's specific isolated database!
-      Using.resource(conn.createStatement()) { stmt =>
-        stmt.execute(MockTexeraDB.getDDLScript)
-      }
+        // AutoCommit is TRUE by default, meaning any records inserted in beforeAll()
+        // will be permanently committed to this suite's specific isolated database!
+        Using.resource(conn.createStatement()) { stmt =>
+          stmt.execute(MockTexeraDB.getDDLScript)
+        }
 
-      connection = Some(conn)
-      val scopedCtx = DSL.using(conn, SQLDialect.POSTGRES)
-      testScopedContext = Some(scopedCtx)
+        connection = Some(conn)
+        val scopedCtx = DSL.using(conn, SQLDialect.POSTGRES)
+        testScopedContext = Some(scopedCtx)
 
-      // Point the Texera backend exactly to this suite's isolated database
-      SqlServer.initConnection(embedded.getJdbcUrl("postgres", uniqueDbName), "postgres", "")
-      SqlServer.getInstance().replaceDSLContext(scopedCtx)
+        // Point the Texera backend exactly to this suite's isolated database
+        SqlServer.initConnection(embedded.getJdbcUrl("postgres", uniqueDbName), "postgres", "")
+        SqlServer.getInstance().replaceDSLContext(scopedCtx)
+      }
     }
-  }
 
   override def withFixture(test: NoArgTest): Outcome = {
     initializeDBAndReplaceDSLContext()
@@ -152,27 +157,29 @@ trait MockTexeraDB extends AnyFlatSpecLike {
     }
   }
 
-  def getDSLContext: DSLContext = synchronized {
-    if (testScopedContext.isEmpty) {
-      initializeDBAndReplaceDSLContext()
+  def getDSLContext: DSLContext =
+    synchronized {
+      if (testScopedContext.isEmpty) {
+        initializeDBAndReplaceDSLContext()
+      }
+      testScopedContext.get
     }
-    testScopedContext.get
-  }
 
   def getDBInstance: EmbeddedPostgres = MockTexeraDB.getDBInstance
 
-  def shutdownDB(): Unit = synchronized {
-    try {
-      connection.foreach { conn =>
-        if (!conn.isClosed) {
-          conn.close()
+  def shutdownDB(): Unit =
+    synchronized {
+      try {
+        connection.foreach { conn =>
+          if (!conn.isClosed) {
+            conn.close()
+          }
         }
+      } catch {
+        case e: Exception => e.printStackTrace()
+      } finally {
+        connection = None
+        testScopedContext = None
       }
-    } catch {
-      case e: Exception => e.printStackTrace()
-    } finally {
-      connection = None
-      testScopedContext = None
     }
-  }
-}
\ No newline at end of file
+}

From dee9df1826c6a09fb329bb32b6fca254ca567e13 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Tue, 7 Jul 2026 14:45:18 -0700
Subject: [PATCH 10/13] style(backend): run scalafmt to format Scala files

---
 .../scala/org/apache/texera/amber/engine/e2e/TestUtils.scala     | 1 -
 1 file changed, 1 deletion(-)

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 b71ac0e749c..eae8bffdb3f 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

From ecdcb92600988dda6b20977e3537ac3d7da6e492 Mon Sep 17 00:00:00 2001
From: Neil Ketteringham <53205839+Neilk1021@users.noreply.github.com>
Date: Wed, 8 Jul 2026 11:05:10 -0700
Subject: [PATCH 11/13] fix(dao): resolve partial initialization in
 MockTexeraDB

Adds proper guards to MockTexeraDB initialization to guard against the DDL script not being loaded.

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Signed-off-by: Neil Ketteringham <53205839+Neilk1021@users.noreply.github.com>
---
 .../scala/org/apache/texera/dao/MockTexeraDB.scala  | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

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 dce303611e8..5d80027cb45 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
@@ -45,14 +45,15 @@ object MockTexeraDB {
 
   def ensureInitialized(): Unit =
     synchronized {
-      if (dbInstance.isDefined) return
+      if (dbInstance.isDefined && ddlScript.isDefined) return
 
-      val driver = new org.postgresql.Driver()
-      DriverManager.registerDriver(driver)
+      if (dbInstance.isEmpty) {
+        val driver = new org.postgresql.Driver()
+        DriverManager.registerDriver(driver)
 
-      // Boot the heavy JVM engine exactly once
-      val embedded = EmbeddedPostgres.builder().start()
-      dbInstance = Some(embedded)
+        // 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)

From a031af64b0447162f3c209494b328c37f0e06e05 Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Wed, 8 Jul 2026 11:23:23 -0700
Subject: [PATCH 12/13] fix (dao): rollback guard in withFixture to prevent
 leakage in the case where a test disables autocommit.

---
 .../src/test/scala/org/apache/texera/dao/MockTexeraDB.scala  | 5 +++++
 1 file changed, 5 insertions(+)

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 5d80027cb45..07d4962a821 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
@@ -138,6 +138,11 @@ trait MockTexeraDB extends AnyFlatSpecLike {
     } finally {
       try {
         if (!conn.isClosed) {
+          if (!conn.getAutoCommit) {
+            conn.rollback()
+            conn.setAutoCommit(true)
+          }
+
           scala.util.Using.resource(conn.createStatement()) { stmt =>
             stmt.execute(
               """

From 642ea2ba533f6817eb040ecc0ed5b564532e22ba Mon Sep 17 00:00:00 2001
From: Neilk1021 <neilk1021@gmail.com>
Date: Wed, 8 Jul 2026 12:37:21 -0700
Subject: [PATCH 13/13] fix (dao): rollback WithFixture to lighter version and
 replacce MockTexeraDB's single connection with Hikari Connection pooling

---
 .../org/apache/texera/dao/SqlServer.scala     | 36 ++-------
 .../org/apache/texera/dao/MockTexeraDB.scala  | 78 +++++++++----------
 2 files changed, 45 insertions(+), 69 deletions(-)

diff --git a/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala b/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
index 9192804efc1..6348ae41fa0 100644
--- a/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
+++ b/common/dao/src/main/scala/org/apache/texera/dao/SqlServer.scala
@@ -94,37 +94,13 @@ object SqlServer {
     * @return
     */
   def withTransaction[T](dsl: DSLContext)(block: DSLContext => T): T = {
-    val provider = dsl.configuration().connectionProvider()
-    val conn = provider.acquire()
-    val originalAutoCommit = conn.getAutoCommit
+    var result: Option[T] = None
 
-    try {
-      if (originalAutoCommit) {
-        conn.setAutoCommit(false)
-      }
+    dsl.transaction(configuration => {
+      val ctx = DSL.using(configuration)
+      result = Some(block(ctx))
+    })
 
-      val txCtx = org.jooq.impl.DSL.using(conn, dsl.dialect())
-      val result = block(txCtx)
-
-      if (originalAutoCommit) {
-        conn.commit()
-      }
-
-      result
-    } catch {
-      case e: Throwable =>
-        if (originalAutoCommit) {
-          conn.rollback()
-        }
-        throw e
-    } finally {
-      try {
-        if (originalAutoCommit != conn.getAutoCommit) {
-          conn.setAutoCommit(originalAutoCommit)
-        }
-      } finally {
-        provider.release(conn)
-      }
-    }
+    result.getOrElse(throw new RuntimeException("Transaction failed without result!"))
   }
 }
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 07d4962a821..bb6791c5f28 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,14 +19,14 @@
 
 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
-import org.scalatest.flatspec.AnyFlatSpecLike
+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
 
@@ -87,14 +87,14 @@ object MockTexeraDB {
   def getDDLScript: String = ddlScript.getOrElse(throw new RuntimeException("DDL not loaded"))
 }
 
-trait MockTexeraDB extends AnyFlatSpecLike {
+trait MockTexeraDB extends TestSuiteMixin { this: TestSuite =>
   private var testScopedContext: Option[DSLContext] = None
-  protected var connection: Option[Connection] = None
+  protected var dataSource: Option[HikariDataSource] = None
   protected var uniqueDbName: String = ""
 
   def initializeDBAndReplaceDSLContext(): Unit =
     synchronized {
-      if (connection.isEmpty || connection.get.isClosed) {
+      if (dataSource.isEmpty || dataSource.get.isClosed) {
         MockTexeraDB.ensureInitialized()
         val embedded = MockTexeraDB.getDBInstance
 
@@ -105,45 +105,54 @@ trait MockTexeraDB extends AnyFlatSpecLike {
           }
         }
 
-        val conn = embedded.getDatabase("postgres", uniqueDbName).getConnection
-
-        // AutoCommit is TRUE by default, meaning any records inserted in beforeAll()
-        // will be permanently committed to this suite's specific isolated database!
-        Using.resource(conn.createStatement()) { stmt =>
-          stmt.execute(MockTexeraDB.getDDLScript)
+        // 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)
+          }
         }
 
-        connection = Some(conn)
-        val scopedCtx = DSL.using(conn, SQLDialect.POSTGRES)
+        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(embedded.getJdbcUrl("postgres", uniqueDbName), "postgres", "")
+        SqlServer.initConnection(jdbcUrl, "postgres", "")
         SqlServer.getInstance().replaceDSLContext(scopedCtx)
       }
     }
 
-  override def withFixture(test: NoArgTest): Outcome = {
+  abstract override def withFixture(test: NoArgTest): Outcome = {
     initializeDBAndReplaceDSLContext()
 
-    val conn = connection.get
     val sqlServerInstance = SqlServer.getInstance()
     val activeContext = testScopedContext.get
 
-    conn.setAutoCommit(true)
-
     try {
       sqlServerInstance.replaceDSLContext(activeContext)
       super.withFixture(test)
     } finally {
       try {
-        if (!conn.isClosed) {
-          if (!conn.getAutoCommit) {
-            conn.rollback()
-            conn.setAutoCommit(true)
-          }
-
-          scala.util.Using.resource(conn.createStatement()) { stmt =>
+        // 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
@@ -175,17 +184,8 @@ trait MockTexeraDB extends AnyFlatSpecLike {
 
   def shutdownDB(): Unit =
     synchronized {
-      try {
-        connection.foreach { conn =>
-          if (!conn.isClosed) {
-            conn.close()
-          }
-        }
-      } catch {
-        case e: Exception => e.printStackTrace()
-      } finally {
-        connection = None
-        testScopedContext = None
-      }
+      try dataSource.foreach(ds => if (!ds.isClosed) ds.close())
+      catch { case e: Exception => e.printStackTrace() }
+      finally { dataSource = None; testScopedContext = None }
     }
 }