[SPARK-42692][CONNECT] Implement Dataset.toJSON

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -29,6 +29,7 @@ import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{PrimitiveLongEnc
 import org.apache.spark.sql.catalyst.expressions.RowOrdering
 import org.apache.spark.sql.connect.client.SparkResult
 import org.apache.spark.sql.connect.common.DataTypeProtoConverter
+import org.apache.spark.sql.functions.{struct, to_json}
 import org.apache.spark.sql.types.{Metadata, StructType}
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.util.Utils
@@ -2777,7 +2778,11 @@ class Dataset[T] private[sql] (
   }
 
   def toJSON: Dataset[String] = {
-    throw new UnsupportedOperationException("toJSON is not implemented.")
+    sparkSession.newDataset(StringEncoder) { builder =>
+      builder.getProjectBuilder
+        .setInput(plan.getRoot)
+        .addExpressions(to_json(struct(Column("*"))).expr)
+    }
   }
 
   private[sql] def analyze: proto.AnalyzePlanResponse = {

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala

@@ -631,6 +631,19 @@ class ClientE2ETestSuite extends RemoteSparkSession with SQLHelper {
     val otherPlan = spark.sql("select 1")
     assert(plan.sameSemantics(otherPlan))
   }
+
+  test("toJSON") {
+    val expected = Array(
+      "{\"b\":0.0,\"id\":0,\"d\":\"world\",\"a\":0}",
+      "{\"b\":0.1,\"id\":1,\"d\":\"world\",\"a\":1}",
+      "{\"b\":0.2,\"id\":2,\"d\":\"world\",\"a\":0}")
+    val result = spark
+      .range(3)
+      .select(generateMyTypeColumns: _*)
+      .toJSON
+      .collect()
+    assert(result sameElements expected)
+  }
 }
 
 private[sql] case class MyType(id: Long, a: Double, b: Double)

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala

@@ -257,6 +257,10 @@ class PlanGenerationTestSuite
     session.read.json(testDataPath.resolve("people.json").toString)
   }
 
+  test("toJSON") {
+    complex.toJSON
+  }
+
   test("read csv") {
     session.read.csv(testDataPath.resolve("people.csv").toString)
   }

connector/connect/common/src/test/resources/query-tests/explain-results/toJSON.explain

@@ -0,0 +1,2 @@
+Project [to_json(struct(id, id#0L, a, a#0, b, b#0, d, d#0, e, e#0, f, f#0, g, g#0), Some(America/Los_Angeles)) AS to_json(struct(id, a, b, d, e, f, g))#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/queries/toJSON.json

@@ -0,0 +1,29 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "project": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "struct\u003cid:bigint,a:int,b:double,d:struct\u003cid:bigint,a:int,b:double\u003e,e:array\u003cint\u003e,f:map\u003cstring,struct\u003cid:bigint,a:int,b:double\u003e\u003e,g:string\u003e"
+      }
+    },
+    "expressions": [{
+      "unresolvedFunction": {
+        "functionName": "to_json",
+        "arguments": [{
+          "unresolvedFunction": {
+            "functionName": "struct",
+            "arguments": [{
+              "unresolvedStar": {
+              }
+            }]
+          }
+        }]
+      }
+    }]
+  }
+}