[SPARK-56388][CONNECT] Add XML support to Spark Connect Parse protocol

python/pyspark/sql/connect/proto/relations_pb2.pyi

@@ -4416,11 +4416,13 @@ class Parse(google.protobuf.message.Message):
         PARSE_FORMAT_UNSPECIFIED: Parse._ParseFormat.ValueType  # 0
         PARSE_FORMAT_CSV: Parse._ParseFormat.ValueType  # 1
         PARSE_FORMAT_JSON: Parse._ParseFormat.ValueType  # 2
+        PARSE_FORMAT_XML: Parse._ParseFormat.ValueType  # 3
 
     class ParseFormat(_ParseFormat, metaclass=_ParseFormatEnumTypeWrapper): ...
     PARSE_FORMAT_UNSPECIFIED: Parse.ParseFormat.ValueType  # 0
     PARSE_FORMAT_CSV: Parse.ParseFormat.ValueType  # 1
     PARSE_FORMAT_JSON: Parse.ParseFormat.ValueType  # 2
+    PARSE_FORMAT_XML: Parse.ParseFormat.ValueType  # 3
 
     class OptionsEntry(google.protobuf.message.Message):
         DESCRIPTOR: google.protobuf.descriptor.Descriptor

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

@@ -377,6 +377,13 @@ class PlanGenerationTestSuite extends ConnectFunSuite with Logging {
       .csv(session.emptyDataset(StringEncoder))
   }
 
+  test("xml from dataset") {
+    session.read
+      .schema(new StructType().add("c1", StringType).add("c2", IntegerType))
+      .option("rowTag", "ROW")
+      .xml(session.emptyDataset(StringEncoder))
+  }
+
   test("read parquet") {
     session.read.parquet(testDataPath.resolve("users.parquet").toString)
   }

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

@@ -1274,6 +1274,38 @@ class ClientE2ETestSuite
     assert(message.contains("PARSE_SYNTAX_ERROR"))
   }
 
+  test("xml from Dataset[String] inferSchema") {
+    val session = spark
+    import session.implicits._
+    val expected = Seq(
+      new GenericRowWithSchema(
+        Array("Ford", "E350", 1997),
+        new StructType()
+          .add("make", StringType)
+          .add("model", StringType)
+          .add("year", LongType)))
+    val ds = Seq("<ROW><year>1997</year><make>Ford</make><model>E350</model></ROW>").toDS()
+    val result = spark.read.option("rowTag", "ROW").xml(ds)
+    checkSameResult(expected, result)
+  }
+
+  test("xml from Dataset[String] with schema") {
+    val session = spark
+    import session.implicits._
+    val schema = new StructType().add("make", StringType).add("model", StringType)
+    val expected = Seq(new GenericRowWithSchema(Array("Ford", "E350"), schema))
+    val ds = Seq("<ROW><year>1997</year><make>Ford</make><model>E350</model></ROW>").toDS()
+    val result = spark.read.schema(schema).option("rowTag", "ROW").xml(ds)
+    checkSameResult(expected, result)
+  }
+
+  test("xml from Dataset[String] with invalid schema") {
+    val message = intercept[ParseException] {
+      spark.read.schema("123").xml(spark.createDataset(Seq.empty[String])(StringEncoder))
+    }.getMessage
+    assert(message.contains("PARSE_SYNTAX_ERROR"))
+  }
+
   test("Dataset result destructive iterator") {
     // Helper methods for accessing private field `idxToBatches` from SparkResult
     val getResultMap =

sql/connect/common/src/main/protobuf/spark/connect/relations.proto

@@ -1215,6 +1215,7 @@ message Parse {
     PARSE_FORMAT_UNSPECIFIED = 0;
     PARSE_FORMAT_CSV = 1;
     PARSE_FORMAT_JSON = 2;
+    PARSE_FORMAT_XML = 3;
   }
 }
 

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataFrameReader.scala

@@ -170,7 +170,7 @@ class DataFrameReader private[sql] (sparkSession: SparkSession) extends sql.Data
 
   /** @inheritdoc */
   def xml(xmlDataset: sql.Dataset[String]): DataFrame =
-    parse(xmlDataset, ParseFormat.PARSE_FORMAT_UNSPECIFIED)
+    parse(xmlDataset, ParseFormat.PARSE_FORMAT_XML)
 
   /** @inheritdoc */
   override def parquet(path: String): DataFrame = super.parquet(path)

sql/connect/common/src/test/resources/query-tests/queries/xml_from_dataset.json

@@ -0,0 +1,39 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "parse": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "{\"type\":\"struct\",\"fields\":[{\"name\":\"value\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}"
+      }
+    },
+    "format": "PARSE_FORMAT_XML",
+    "schema": {
+      "struct": {
+        "fields": [{
+          "name": "c1",
+          "dataType": {
+            "string": {
+              "collation": "UTF8_BINARY"
+            }
+          },
+          "nullable": true
+        }, {
+          "name": "c2",
+          "dataType": {
+            "integer": {
+            }
+          },
+          "nullable": true
+        }]
+      }
+    },
+    "options": {
+      "rowtag": "ROW"
+    }
+  }
+}

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala

@@ -1767,6 +1767,8 @@ class SparkConnectPlanner(
         dataFrameReader.csv(ds).queryExecution.analyzed
       case ParseFormat.PARSE_FORMAT_JSON =>
         dataFrameReader.json(ds).queryExecution.analyzed
+      case ParseFormat.PARSE_FORMAT_XML =>
+        dataFrameReader.xml(ds).queryExecution.analyzed
       case other => throw InvalidInputErrors.invalidEnum(other)
     }
   }