[SPARK-32524][SQL][TESTS] CachedBatchSerializerSuite should clean up InMemoryRelation.ser

dongjoon-hyun · dongjoon-hyun · commit a5522874e972 · 2020-08-03T22:57:37.000-07:00
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
@@ -277,6 +277,9 @@ object InMemoryRelation {
     ser.get
   }
 
+  /* Visible for testing */
+  private[spark] def clearSerializer(): Unit = synchronized { ser = None }
+
   def convertToColumnarIfPossible(plan: SparkPlan): SparkPlan = plan match {
     case gen: WholeStageCodegenExec => gen.child match {
       case c2r: ColumnarToRowTransition => c2r.child match {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/CachedBatchSerializerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/CachedBatchSerializerSuite.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.{QueryTest, Row}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
 import org.apache.spark.sql.columnar.{CachedBatch, CachedBatchSerializer}
+import org.apache.spark.sql.execution.columnar.InMemoryRelation.clearSerializer
 import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.test.SharedSparkSession
@@ -120,6 +121,11 @@ class CachedBatchSerializerSuite  extends QueryTest with SharedSparkSession {
       classOf[TestSingleIntColumnarCachedBatchSerializer].getName)
   }
 
+  protected override def afterAll(): Unit = {
+    clearSerializer()
+    super.afterAll()
+  }
+
   test("Columnar Cache Plugin") {
     withTempPath { workDir =>
       val workDirPath = workDir.getAbsolutePath

Original file line number	Diff line number	Diff line change
`@@ -277,6 +277,9 @@ object InMemoryRelation {`
`277`	`277`	`ser.get`
`278`	`278`	`}`
`279`	`279`
	`280`	`+ /* Visible for testing */`
	`281`	`+ private[spark] def clearSerializer(): Unit = synchronized { ser = None }`
	`282`	`+`
`280`	`283`	`def convertToColumnarIfPossible(plan: SparkPlan): SparkPlan = plan match {`
`281`	`284`	`case gen: WholeStageCodegenExec => gen.child match {`
`282`	`285`	`case c2r: ColumnarToRowTransition => c2r.child match {`