make toJSON not go through rdd form but operate on dataset always (#48)

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -2631,16 +2631,17 @@ class Dataset[T] private[sql](
    * @since 2.0.0
    */
   def toJSON: Dataset[String] = {
-    val rowSchema = this.schema
-    val rdd: RDD[String] = queryExecution.toRdd.mapPartitions { iter =>
+    val rowSchema = schema
+    import sparkSession.implicits.newStringEncoder
+    mapPartitions { iter =>
       val writer = new CharArrayWriter()
       // create the Generator without separator inserted between 2 records
       val gen = new JacksonGenerator(rowSchema, writer)
 
       new Iterator[String] {
         override def hasNext: Boolean = iter.hasNext
         override def next(): String = {
-          gen.write(iter.next())
+          gen.write(exprEnc.toRow(iter.next()))
           gen.flush()
 
           val json = writer.toString
@@ -2654,8 +2655,6 @@ class Dataset[T] private[sql](
         }
       }
     }
-    import sparkSession.implicits.newStringEncoder
-    sparkSession.createDataset(rdd)
   }
 
   /**

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

@@ -156,6 +156,16 @@ class JsonFileFormat extends TextBasedFileFormat with DataSourceRegister {
   override def equals(other: Any): Boolean = other.isInstanceOf[JsonFileFormat]
 }
 
+/**
+ * Implementation of OutputWriter for Json file format. Uses [[JacksonGenerator]]
+ * to do json conversion.
+ *
+ * @param path output location for the job
+ * @param options json generator options
+ * @param bucketId bucket for this partition
+ * @param dataSchema JacksonGenerator factory
+ * @param context hadoop task context
+ */
 private[json] class JsonOutputWriter(
     path: String,
     options: JSONOptions,