[SPARK-26208][SQL] add headers to empty csv files when header=true

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

@@ -60,6 +60,8 @@ 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,6 +49,8 @@ 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/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityGenerator.scala

@@ -32,7 +32,6 @@ class UnivocityGenerator(
   private val writerSettings = options.asWriterSettings
   writerSettings.setHeaders(schema.fieldNames: _*)
   private val gen = new CsvWriter(writer, writerSettings)
-  private var printHeader = options.headerFlag
 
   // A `ValueConverter` is responsible for converting a value of an `InternalRow` to `String`.
   // When the value is null, this converter should not be called.
@@ -72,15 +71,15 @@ class UnivocityGenerator(
     values
   }
 
+  def writeHeaders(): Unit = {
+    gen.writeHeaders()
+  }
+
   /**
    * Writes a single InternalRow to CSV using Univocity.
    */
   def write(row: InternalRow): Unit = {
-    if (printHeader) {
-      gen.writeHeaders()
-    }
     gen.writeRow(convertRow(row): _*)
-    printHeader = false
   }
 
   def writeToString(row: InternalRow): String = {

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

@@ -121,6 +121,7 @@ class SingleDirectoryDataWriter(
       path = currentPath,
       dataSchema = description.dataColumns.toStructType,
       context = taskAttemptContext)
+    currentWriter.init()
 
     statsTrackers.foreach(_.newFile(currentPath))
   }
@@ -237,6 +238,7 @@ class DynamicPartitionDataWriter(
       path = currentPath,
       dataSchema = description.dataColumns.toStructType,
       context = taskAttemptContext)
+    currentWriter.init()
 
     statsTrackers.foreach(_.newFile(currentPath))
   }

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

@@ -57,6 +57,9 @@ abstract class OutputWriterFactory extends Serializable {
  * executor side.  This instance is used to persist rows to this single output file.
  */
 abstract class OutputWriter {
+  /** Initializes before writing any rows. Invoked on executor size. */
+  def init(): Unit
+
   /**
    * Persists a single row.  Invoked on the executor side.  When writing to dynamically partitioned
    * tables, dynamic partition columns are not included in rows to be written.

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

@@ -175,6 +175,12 @@ private[csv] class CsvOutputWriter(
 
   private val gen = new UnivocityGenerator(dataSchema, writer, params)
 
+  override def init(): Unit = {
+    if (params.headerFlag) {
+      gen.writeHeaders()
+    }
+  }
+
   override def write(row: InternalRow): Unit = gen.write(row)
 
   override def close(): Unit = gen.close()

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

@@ -181,6 +181,8 @@ private[json] class JsonOutputWriter(
   // create the Generator without separator inserted between 2 records
   private[this] val gen = new JacksonGenerator(dataSchema, writer, options)
 
+  override def init(): Unit = {}
+
   override def write(row: InternalRow): Unit = {
     gen.write(row)
     gen.writeLineEnding()

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

@@ -50,6 +50,8 @@ 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,6 +37,8 @@ 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

@@ -150,6 +150,8 @@ class TextOutputWriter(
 
   private val writer = CodecStreams.createOutputStream(context, new Path(path))
 
+  override def init(): Unit = {}
+
   override def write(row: InternalRow): Unit = {
     if (!row.isNullAt(0)) {
       val utf8string = row.getUTF8String(0)

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala

@@ -1986,4 +1986,16 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
     }.getMessage
     assert(errMsg2.contains("'lineSep' can contain only 1 character"))
   }
+
+  test("SPARK-26208: write and read empty data to csv file with header") {
+    withTempPath { path =>
+      val df1 = Seq.empty[(String, String)].toDF("x", "y")
+      df1.printSchema
+      df1.write.format("csv").option("header", true).save(path.getAbsolutePath)
+      val df2 = spark.read.format("csv").option("header", true).load(path.getAbsolutePath)
+      df2.printSchema
+      assert(df1.schema === df2.schema)
+      checkAnswer(df1, df2)
+    }
+  }
 }

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

@@ -141,6 +141,8 @@ 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,6 +272,8 @@ 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,6 +122,8 @@ 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