allow empty batch

Author: jose-torres

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/ConsoleWriter.scala

@@ -38,7 +38,7 @@ class ConsoleWriter(batchId: Long, schema: StructType, options: Map[String, Stri
   override def commit(messages: Array[WriterCommitMessage]): Unit = synchronized {
     val batch = messages.collect {
       case PackedRowCommitMessage(rows) => rows
-    }.reduce(_ ++ _)
+    }.fold(Array())(_ ++ _)
 
     // scalastyle:off println
     println("-------------------------------------------")

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/sources/ConsoleWriterSuite.scala

@@ -36,6 +36,8 @@ class ConsoleWriterSuite extends StreamTest {
         query.processAllAvailable()
         input.addData(4, 5, 6)
         query.processAllAvailable()
+        input.addData()
+        query.processAllAvailable()
       } finally {
         query.stop()
       }
@@ -64,6 +66,14 @@ class ConsoleWriterSuite extends StreamTest {
         ||    6|
         |+-----+
         |
+        |-------------------------------------------
+        |Batch: 2
+        |-------------------------------------------
+        |+-----+
+        ||value|
+        |+-----+
+        |+-----+
+        |
         |""".stripMargin)
   }
 }