revert

Author: cloud-fan

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala

@@ -169,7 +169,6 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
     log.debug(
       s"Creating MutableProj: $expressions, inputSchema: $inputSchema, codegen:$codegenEnabled")
     if(codegenEnabled && expressions.forall(_.isThreadSafe)) {
-
       GenerateMutableProjection.generate(expressions, inputSchema)
     } else {
       () => new InterpretedMutableProjection(expressions, inputSchema)

sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala

@@ -166,7 +166,8 @@ case class TakeOrderedAndProject(
 
   private val ord: RowOrdering = new RowOrdering(sortOrder, child.output)
 
-  private val projection = projectList.map(newProjection(_, child.output))
+  // TODO: remove @transient after figure out how to clean closure at InsertIntoHiveTable.
+  @transient private val projection = projectList.map(new InterpretedProjection(_, child.output))
 
   private def collectData(): Array[InternalRow] = {
     val data = child.execute().map(_.copy()).takeOrdered(limit)(ord)

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

@@ -48,16 +48,16 @@ case class InsertIntoHiveTable(
     overwrite: Boolean,
     ifNotExists: Boolean) extends UnaryNode with HiveInspectors {
 
-  val sc: HiveContext = sqlContext.asInstanceOf[HiveContext]
-  lazy val outputClass = newSerializer(table.tableDesc).getSerializedClass
-  private lazy val hiveContext = new Context(sc.hiveconf)
-  private lazy val catalog = sc.catalog
+  @transient val sc: HiveContext = sqlContext.asInstanceOf[HiveContext]
+  @transient lazy val outputClass = newSerializer(table.tableDesc).getSerializedClass
+  @transient private lazy val hiveContext = new Context(sc.hiveconf)
+  @transient private lazy val catalog = sc.catalog
 
-  private val newSerializer = (tableDesc: TableDesc) => {
+  private def newSerializer(tableDesc: TableDesc): Serializer = {
     val serializer = tableDesc.getDeserializerClass.newInstance().asInstanceOf[Serializer]
     serializer.initialize(null, tableDesc.getProperties)
     serializer
-  }: Serializer
+  }
 
   def output: Seq[Attribute] = child.output
 
@@ -79,10 +79,13 @@ case class InsertIntoHiveTable(
       SparkHiveWriterContainer.createPathFromString(fileSinkConf.getDirName, conf.value))
     log.debug("Saving as hadoop file of type " + valueClass.getSimpleName)
 
-    val newSer = newSerializer
-    val schema = table.schema
+    writerContainer.driverSideSetup()
+    sc.sparkContext.runJob(rdd, writeToFile _)
+    writerContainer.commitJob()
+
     // Note that this function is executed on executor side
-    val writeToFile = (context: TaskContext, iterator: Iterator[InternalRow]) => {
+    def writeToFile(context: TaskContext, iterator: Iterator[InternalRow]): Unit = {
+      val serializer = newSerializer(fileSinkConf.getTableInfo)
       val standardOI = ObjectInspectorUtils
         .getStandardObjectInspector(
           fileSinkConf.getTableInfo.getDeserializer.getObjectInspector,
@@ -103,16 +106,12 @@ case class InsertIntoHiveTable(
         }
 
         writerContainer
-          .getLocalFileWriter(row, schema)
-          .write(newSer(fileSinkConf.getTableInfo).serialize(outputData, standardOI))
+          .getLocalFileWriter(row, table.schema)
+          .write(serializer.serialize(outputData, standardOI))
       }
 
       writerContainer.close()
-    }: Unit
-
-    writerContainer.driverSideSetup()
-    sc.sparkContext.runJob(rdd, sc.sparkContext.clean(writeToFile))
-    writerContainer.commitJob()
+    }
   }
 
   /**