provide default do-nothing implementation for OutputWriter.init

Author: koertkuipers

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala

@@ -60,8 +60,6 @@ private[avro] class AvroOutputWriter(
 
     }.getRecordWriter(context)
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     val key = new AvroKey(serializer.serialize(row).asInstanceOf[GenericRecord])
     recordWriter.write(key, NullWritable.get())

mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala

@@ -49,8 +49,6 @@ private[libsvm] class LibSVMOutputWriter(
   // This `asInstanceOf` is safe because it's guaranteed by `LibSVMFileFormat.verifySchema`
   private val udt = dataSchema(1).dataType.asInstanceOf[VectorUDT]
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     val label = row.getDouble(0)
     val vector = udt.deserialize(row.getStruct(1, udt.sqlType.length))

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/OutputWriter.scala

@@ -58,7 +58,7 @@ abstract class OutputWriterFactory extends Serializable {
  */
 abstract class OutputWriter {
   /** Initializes before writing any rows. Invoked on executor size. */
-  def init(): Unit
+  def init(): Unit = {}
 
   /**
    * Persists a single row.  Invoked on the executor side.  When writing to dynamically partitioned

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala

@@ -177,8 +177,6 @@ private[json] class JsonOutputWriter(
 
   private var jacksonGenerator: Option[JacksonGenerator] = None
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     val gen = jacksonGenerator.getOrElse {
       val os = CodecStreams.createOutputStreamWriter(context, new Path(path), encoding)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcOutputWriter.scala

@@ -50,8 +50,6 @@ private[orc] class OrcOutputWriter(
     recordWriter
   }
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     recordWriter.write(NullWritable.get(), serializer.serialize(row))
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetOutputWriter.scala

@@ -37,8 +37,6 @@ private[parquet] class ParquetOutputWriter(path: String, context: TaskAttemptCon
     }.getRecordWriter(context)
   }
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = recordWriter.write(null, row)
 
   override def close(): Unit = recordWriter.close(context)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala

@@ -152,8 +152,6 @@ class TextOutputWriter(
 
   private var outputStream: Option[OutputStream] = None
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     val os = outputStream.getOrElse{
       val newStream = CodecStreams.createOutputStream(context, new Path(path))

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

@@ -141,8 +141,6 @@ class HiveOutputWriter(
   private val wrappers = fieldOIs.zip(dataTypes).map { case (f, dt) => wrapperFor(f, dt) }
   private val outputData = new Array[Any](fieldOIs.length)
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     var i = 0
     while (i < fieldOIs.length) {

sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala

@@ -272,8 +272,6 @@ private[orc] class OrcOutputWriter(
     ).asInstanceOf[RecordWriter[NullWritable, Writable]]
   }
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     recordWriter.write(NullWritable.get(), serializer.serialize(row))
   }

sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextRelation.scala

@@ -122,8 +122,6 @@ class SimpleTextOutputWriter(path: String, dataSchema: StructType, context: Task
 
   private val writer = CodecStreams.createOutputStreamWriter(context, new Path(path))
 
-  override def init(): Unit = {}
-
   override def write(row: InternalRow): Unit = {
     val serialized = row.toSeq(dataSchema).map { v =>
       if (v == null) "" else v.toString