Isolate cached serializers between SQL test suites [fast-ut] [reduced-it]

tests/src/test/scala/com/nvidia/spark/rapids/ParallelUnitTestRunner.scala

@@ -597,7 +597,7 @@ object ParallelUnitTestRunner {
       }
     }
     cleanup(cleanupSparkSessionAndContext())
-    cleanup(clearCachedBatchSerializer())
+    cleanup(TestUtils.clearCachedBatchSerializer())
     cleanup {
       warehouseDirs ++= Option(tmpDir.toFile.listFiles()).getOrElse(Array.empty[File])
           .filter(file => file.isDirectory && file.getName.startsWith(SPARK_WAREHOUSE_PREFIX))
@@ -626,25 +626,6 @@ object ParallelUnitTestRunner {
     SparkSession.clearDefaultSession()
   }
 
-  /**
-   * Reset Spark's JVM-global cached CachedBatchSerializer between suites.
-   *
-   * The first time any suite builds a cached relation, Spark's InMemoryRelation caches the
-   * serializer named by spark.sql.cache.serializer in a JVM-global field and reuses it for the
-   * lifetime of the JVM, ignoring the configuration of later sessions. In a persistent worker a
-   * suite that runs without the RAPIDS serializer would pin Spark's default serializer, so a later
-   * suite such as RapidsCachedTableSuite would observe a serializer that no longer matches its own
-   * spark.sql.cache.serializer and fail ("Cache serializer failed to load!"). Clearing the cache
-   * between suites lets each suite rebuild the serializer from its own configuration.
-   *
-   * Called reflectively because InMemoryRelation.clearSerializer is not part of the public API.
-   */
-  private def clearCachedBatchSerializer(): Unit = {
-    val module = Class.forName("org.apache.spark.sql.execution.columnar.InMemoryRelation$")
-        .getField("MODULE$").get(null)
-    module.getClass.getMethod("clearSerializer").invoke(module)
-  }
-
   private def poolWorkerCommand(
       workerId: Int,
       runId: Int,

tests/src/test/scala/com/nvidia/spark/rapids/TestUtils.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2025, NVIDIA CORPORATION.
+ * Copyright (c) 2020-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -44,6 +44,14 @@ object TestUtils extends Assertions {
     System.getProperty("test.build.data", System.getProperty("java.io.tmpdir", "/tmp")),
     basename)
 
+  // Spark caches the configured serializer in a JVM-global singleton, so suites that select a
+  // different serializer must reset it at suite boundaries.
+  def clearCachedBatchSerializer(): Unit = {
+    val module = Class.forName("org.apache.spark.sql.execution.columnar.InMemoryRelation$")
+        .getField("MODULE$").get(null)
+    module.getClass.getMethod("clearSerializer").invoke(module)
+  }
+
   /** Compare the equality of two tables */
   def compareTables(expected: Table, actual: Table): Unit = {
     assertResult(expected.getRowCount)(actual.getRowCount)

tests/src/test/spark330/scala/org/apache/spark/sql/rapids/utils/RapidsSQLTestsBaseTrait.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024-2025, NVIDIA CORPORATION.
+ * Copyright (c) 2024-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ package org.apache.spark.sql.rapids.utils
 
 import java.util.{Locale, TimeZone}
 
+import com.nvidia.spark.rapids.TestUtils
 import org.apache.hadoop.fs.FileUtil
 import org.scalactic.source.Position
 import org.scalatest.Tag
@@ -38,10 +39,17 @@ import org.apache.spark.sql.test.SharedSparkSession
 
 /** Basic trait for Rapids SQL test cases. */
 trait RapidsSQLTestsBaseTrait extends SharedSparkSession with RapidsTestsBaseTrait {
-  protected override def afterAll(): Unit = {
+  protected override def beforeAll(): Unit = {
+    TestUtils.clearCachedBatchSerializer()
+    super.beforeAll()
+  }
+
+  protected override def afterAll(): Unit = try {
     // SparkFunSuite will set this to true, and forget to reset to false
     System.clearProperty(IS_TESTING.key)
     super.afterAll()
+  } finally {
+    TestUtils.clearCachedBatchSerializer()
   }
 
   override protected def testFile(fileName: String): String = {