save

Author: AngersZhuuuu

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

@@ -89,15 +89,23 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
   }
 }
 
-abstract class BaseScriptTransformationWriterThread(
-    iter: Iterator[InternalRow],
-    inputSchema: Seq[DataType],
-    ioSchema: BaseScriptTransformIOSchema,
-    outputStream: OutputStream,
-    proc: Process,
-    stderrBuffer: CircularBuffer,
-    taskContext: TaskContext,
-    conf: Configuration) extends Thread with Logging {
+abstract class BaseScriptTransformationWriterThread extends Thread with Logging {
+
+  def iter: Iterator[InternalRow]
+
+  def inputSchema: Seq[DataType]
+
+  def ioSchema: BaseScriptTransformIOSchema
+
+  def outputStream: OutputStream
+
+  def proc: Process
+
+  def stderrBuffer: CircularBuffer
+
+  def taskContext: TaskContext
+
+  def conf: Configuration
 
   setDaemon(true)
 

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

@@ -71,7 +71,7 @@ case class SparkScriptTransformationExec(
 
     // This new thread will consume the ScriptTransformation's input rows and write them to the
     // external process. That process's output will be read by this current thread.
-    val writerThread = new ScriptTransformationWriterThread(
+    val writerThread = ScriptTransformationWriterThread(
       inputIterator.map(outputProjection),
       input.map(_.dataType),
       ioschema,
@@ -131,7 +131,7 @@ case class SparkScriptTransformationExec(
   }
 }
 
-private class ScriptTransformationWriterThread(
+case class ScriptTransformationWriterThread(
     iter: Iterator[InternalRow],
     inputSchema: Seq[DataType],
     ioSchema: SparkScriptIOSchema,
@@ -140,15 +140,7 @@ private class ScriptTransformationWriterThread(
     stderrBuffer: CircularBuffer,
     taskContext: TaskContext,
     conf: Configuration)
-  extends BaseScriptTransformationWriterThread(
-    iter,
-    inputSchema,
-    ioSchema,
-    outputStream,
-    proc,
-    stderrBuffer,
-    taskContext,
-    conf) {
+  extends BaseScriptTransformationWriterThread {
 
   setDaemon(true)
 

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

@@ -86,7 +86,7 @@ case class HiveScriptTransformationExec(
 
     // This new thread will consume the ScriptTransformation's input rows and write them to the
     // external process. That process's output will be read by this current thread.
-    val writerThread = new HiveScriptTransformationWriterThread(
+    val writerThread = HiveScriptTransformationWriterThread(
       inputIterator.map(outputProjection),
       input.map(_.dataType),
       inputSerde,
@@ -208,7 +208,7 @@ case class HiveScriptTransformationExec(
   }
 }
 
-private class HiveScriptTransformationWriterThread(
+case class HiveScriptTransformationWriterThread(
     iter: Iterator[InternalRow],
     inputSchema: Seq[DataType],
     @Nullable inputSerde: AbstractSerDe,
@@ -219,15 +219,7 @@ private class HiveScriptTransformationWriterThread(
     stderrBuffer: CircularBuffer,
     taskContext: TaskContext,
     conf: Configuration)
-  extends BaseScriptTransformationWriterThread(
-    iter,
-    inputSchema,
-    ioSchema,
-    outputStream,
-    proc,
-    stderrBuffer,
-    taskContext,
-    conf) with HiveInspectors {
+  extends BaseScriptTransformationWriterThread with HiveInspectors {
 
   override def processRows(): Unit = {
     val dataOutputStream = new DataOutputStream(outputStream)