[SPARK-52448][CONNECT] Add simplified Struct Expression.Literal

### What changes were proposed in this pull request?

This PR adds a new `data_type_struct` field to the protobuf definition for struct literals in Spark Connect, addressing the ambiguity issues with the existing `struct_type` field. The changes include:

1. **Protobuf Schema Update**: Added a new `data_type_struct` field of type `DataType.Struct` to the `Literal.Struct` message in `expressions.proto`, while marking the existing `struct_type` field as deprecated.

2. **Enhanced Struct Conversion Logic**: Updated `LiteralValueProtoConverter.scala` to:
   - Use the new `data_type_struct` field when available for more precise struct type definition
   - Maintain backward compatibility by still supporting the deprecated `struct_type` field
   - Add proper field metadata handling in struct conversions
   - Improve type inference for struct fields when data types can be inferred from literal values

### Why are the changes needed?

The current Expression.Struct literal is somewhat overcomplicated since it duplicates most of the information its fields already have. This is bulky to send over the wire, and it can be ambiguous.

### Does this PR introduce _any_ user-facing change?

No. This PR maintains backward compatibility with existing struct literal implementations. Existing code using the deprecated `struct_type` field will continue to work without modification.

### How was this patch tested?

`build/sbt "connect/testOnly *LiteralExpressionProtoConverterSuite"`

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Cursor 1.2.4

Closes apache#51561 from heyihong/SPARK-52448.

Authored-by: Yihong He <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: heyihong

python/pyspark/sql/connect/proto/expressions_pb2.py

@@ -554,27 +554,51 @@ class Expression(google.protobuf.message.Message):
 
             STRUCT_TYPE_FIELD_NUMBER: builtins.int
             ELEMENTS_FIELD_NUMBER: builtins.int
+            DATA_TYPE_STRUCT_FIELD_NUMBER: builtins.int
             @property
-            def struct_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType: ...
+            def struct_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType:
+                """(Deprecated) The type of the struct.
+
+                This field is deprecated since Spark 4.1+ because using DataType as the type of a struct
+                is ambiguous. This field should only be set if the data_type_struct field is not set.
+                Use data_type_struct field instead.
+                """
             @property
             def elements(
                 self,
             ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
                 global___Expression.Literal
-            ]: ...
+            ]:
+                """(Required) The literal values that make up the struct elements."""
+            @property
+            def data_type_struct(self) -> pyspark.sql.connect.proto.types_pb2.DataType.Struct:
+                """The type of the struct.
+
+                Whether data_type_struct.fields.data_type should be set depends on
+                whether each field's type can be inferred from the elements field.
+                """
             def __init__(
                 self,
                 *,
                 struct_type: pyspark.sql.connect.proto.types_pb2.DataType | None = ...,
                 elements: collections.abc.Iterable[global___Expression.Literal] | None = ...,
+                data_type_struct: pyspark.sql.connect.proto.types_pb2.DataType.Struct | None = ...,
             ) -> None: ...
             def HasField(
-                self, field_name: typing_extensions.Literal["struct_type", b"struct_type"]
+                self,
+                field_name: typing_extensions.Literal[
+                    "data_type_struct", b"data_type_struct", "struct_type", b"struct_type"
+                ],
             ) -> builtins.bool: ...
             def ClearField(
                 self,
                 field_name: typing_extensions.Literal[
-                    "elements", b"elements", "struct_type", b"struct_type"
+                    "data_type_struct",
+                    b"data_type_struct",
+                    "elements",
+                    b"elements",
+                    "struct_type",
+                    b"struct_type",
                 ],
             ) -> None: ...
 

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

@@ -79,15 +79,24 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
       Literal((12.0, "north", 60.0, "west"), Option(dataType)),
       expr { b =>
         val builder = b.getLiteralBuilder.getStructBuilder
-        builder.getStructTypeBuilder.getStructBuilder
-          .addFields(structField("_1", ProtoDataTypes.DoubleType))
-          .addFields(structField("_2", stringTypeWithCollation))
-          .addFields(structField("_3", ProtoDataTypes.DoubleType))
-          .addFields(structField("_4", stringTypeWithCollation))
-        builder.addElements(proto.Expression.Literal.newBuilder().setDouble(12.0))
-        builder.addElements(proto.Expression.Literal.newBuilder().setString("north"))
-        builder.addElements(proto.Expression.Literal.newBuilder().setDouble(60.0))
-        builder.addElements(proto.Expression.Literal.newBuilder().setString("west"))
+        builder
+          .addElements(proto.Expression.Literal.newBuilder().setDouble(12.0).build())
+        builder
+          .addElements(proto.Expression.Literal.newBuilder().setString("north").build())
+        builder
+          .addElements(proto.Expression.Literal.newBuilder().setDouble(60.0).build())
+        builder
+          .addElements(proto.Expression.Literal.newBuilder().setString("west").build())
+        builder.setDataTypeStruct(
+          proto.DataType.Struct
+            .newBuilder()
+            .addFields(
+              proto.DataType.StructField.newBuilder().setName("_1").setNullable(true).build())
+            .addFields(structField("_2", stringTypeWithCollation))
+            .addFields(
+              proto.DataType.StructField.newBuilder().setName("_3").setNullable(true).build())
+            .addFields(structField("_4", stringTypeWithCollation))
+            .build())
       })
   }
 

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

@@ -227,8 +227,21 @@ message Expression {
     }
 
     message Struct {
-      DataType struct_type = 1;
+      // (Deprecated) The type of the struct.
+      //
+      // This field is deprecated since Spark 4.1+ because using DataType as the type of a struct
+      // is ambiguous. This field should only be set if the data_type_struct field is not set.
+      // Use data_type_struct field instead.
+      DataType struct_type = 1 [deprecated = true];
+
+      // (Required) The literal values that make up the struct elements.
       repeated Literal elements = 2;
+
+      // The type of the struct.
+      //
+      // Whether data_type_struct.fields.data_type should be set depends on
+      // whether each field's type can be inferred from the elements field.
+      DataType.Struct data_type_struct = 3;
     }
 
     message SpecializedArray {

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

@@ -149,17 +149,36 @@ object LiteralValueProtoConverter {
     }
 
     def structBuilder(scalaValue: Any, structType: StructType) = {
-      val sb = builder.getStructBuilder.setStructType(toConnectProtoType(structType))
-      val dataTypes = structType.fields.map(_.dataType)
+      val sb = builder.getStructBuilder
+      val fields = structType.fields
 
       scalaValue match {
         case p: Product =>
           val iter = p.productIterator
+          val dataTypeStruct = proto.DataType.Struct.newBuilder()
           var idx = 0
           while (idx < structType.size) {
-            sb.addElements(toLiteralProto(iter.next(), dataTypes(idx)))
+            val field = fields(idx)
+            val literalProto = toLiteralProto(iter.next(), field.dataType)
+            sb.addElements(literalProto)
+
+            val fieldBuilder = dataTypeStruct
+              .addFieldsBuilder()
+              .setName(field.name)
+              .setNullable(field.nullable)
+
+            if (LiteralValueProtoConverter.getInferredDataType(literalProto).isEmpty) {
+              fieldBuilder.setDataType(toConnectProtoType(field.dataType))
+            }
+
+            // Set metadata if available
+            if (field.metadata != Metadata.empty) {
+              fieldBuilder.setMetadata(field.metadata.json)
+            }
+
             idx += 1
           }
+          sb.setDataTypeStruct(dataTypeStruct.build())
         case other =>
           throw new IllegalArgumentException(s"literal $other not supported (yet).")
       }
@@ -300,54 +319,101 @@ object LiteralValueProtoConverter {
       case proto.Expression.Literal.LiteralTypeCase.ARRAY =>
         toCatalystArray(literal.getArray)
 
+      case proto.Expression.Literal.LiteralTypeCase.STRUCT =>
+        toCatalystStruct(literal.getStruct)._1
+
       case other =>
         throw new UnsupportedOperationException(
           s"Unsupported Literal Type: ${other.getNumber} (${other.name})")
     }
   }
 
-  private def getConverter(dataType: proto.DataType): proto.Expression.Literal => Any = {
-    if (dataType.hasShort) { v =>
-      v.getShort.toShort
-    } else if (dataType.hasInteger) { v =>
-      v.getInteger
-    } else if (dataType.hasLong) { v =>
-      v.getLong
-    } else if (dataType.hasDouble) { v =>
-      v.getDouble
-    } else if (dataType.hasByte) { v =>
-      v.getByte.toByte
-    } else if (dataType.hasFloat) { v =>
-      v.getFloat
-    } else if (dataType.hasBoolean) { v =>
-      v.getBoolean
-    } else if (dataType.hasString) { v =>
-      v.getString
-    } else if (dataType.hasBinary) { v =>
-      v.getBinary.toByteArray
-    } else if (dataType.hasDate) { v =>
-      v.getDate
-    } else if (dataType.hasTimestamp) { v =>
-      v.getTimestamp
-    } else if (dataType.hasTimestampNtz) { v =>
-      v.getTimestampNtz
-    } else if (dataType.hasDayTimeInterval) { v =>
-      v.getDayTimeInterval
-    } else if (dataType.hasYearMonthInterval) { v =>
-      v.getYearMonthInterval
-    } else if (dataType.hasDecimal) { v =>
-      Decimal(v.getDecimal.getValue)
-    } else if (dataType.hasCalendarInterval) { v =>
-      val interval = v.getCalendarInterval
-      new CalendarInterval(interval.getMonths, interval.getDays, interval.getMicroseconds)
-    } else if (dataType.hasArray) { v =>
-      toCatalystArray(v.getArray)
-    } else if (dataType.hasMap) { v =>
-      toCatalystMap(v.getMap)
-    } else if (dataType.hasStruct) { v =>
-      toCatalystStruct(v.getStruct)
-    } else {
-      throw InvalidPlanInput(s"Unsupported Literal Type: $dataType)")
+  private def getConverter(
+      dataType: proto.DataType,
+      inferDataType: Boolean = false): proto.Expression.Literal => Any = {
+    dataType.getKindCase match {
+      case proto.DataType.KindCase.SHORT => v => v.getShort.toShort
+      case proto.DataType.KindCase.INTEGER => v => v.getInteger
+      case proto.DataType.KindCase.LONG => v => v.getLong
+      case proto.DataType.KindCase.DOUBLE => v => v.getDouble
+      case proto.DataType.KindCase.BYTE => v => v.getByte.toByte
+      case proto.DataType.KindCase.FLOAT => v => v.getFloat
+      case proto.DataType.KindCase.BOOLEAN => v => v.getBoolean
+      case proto.DataType.KindCase.STRING => v => v.getString
+      case proto.DataType.KindCase.BINARY => v => v.getBinary.toByteArray
+      case proto.DataType.KindCase.DATE => v => v.getDate
+      case proto.DataType.KindCase.TIMESTAMP => v => v.getTimestamp
+      case proto.DataType.KindCase.TIMESTAMP_NTZ => v => v.getTimestampNtz
+      case proto.DataType.KindCase.DAY_TIME_INTERVAL => v => v.getDayTimeInterval
+      case proto.DataType.KindCase.YEAR_MONTH_INTERVAL => v => v.getYearMonthInterval
+      case proto.DataType.KindCase.DECIMAL => v => Decimal(v.getDecimal.getValue)
+      case proto.DataType.KindCase.CALENDAR_INTERVAL =>
+        v =>
+          val interval = v.getCalendarInterval
+          new CalendarInterval(interval.getMonths, interval.getDays, interval.getMicroseconds)
+      case proto.DataType.KindCase.ARRAY => v => toCatalystArray(v.getArray)
+      case proto.DataType.KindCase.MAP => v => toCatalystMap(v.getMap)
+      case proto.DataType.KindCase.STRUCT =>
+        if (inferDataType) { v =>
+          val (struct, structType) = toCatalystStruct(v.getStruct, None)
+          LiteralValueWithDataType(
+            struct,
+            proto.DataType.newBuilder.setStruct(structType).build())
+        } else { v =>
+          toCatalystStruct(v.getStruct, Some(dataType.getStruct))._1
+        }
+      case _ =>
+        throw InvalidPlanInput(s"Unsupported Literal Type: $dataType)")
+    }
+  }
+
+  private def getInferredDataType(literal: proto.Expression.Literal): Option[proto.DataType] = {
+    if (literal.hasNull) {
+      return Some(literal.getNull)
+    }
+
+    val builder = proto.DataType.newBuilder()
+    literal.getLiteralTypeCase match {
+      case proto.Expression.Literal.LiteralTypeCase.BINARY =>
+        builder.setBinary(proto.DataType.Binary.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.BOOLEAN =>
+        builder.setBoolean(proto.DataType.Boolean.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.BYTE =>
+        builder.setByte(proto.DataType.Byte.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.SHORT =>
+        builder.setShort(proto.DataType.Short.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.INTEGER =>
+        builder.setInteger(proto.DataType.Integer.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.LONG =>
+        builder.setLong(proto.DataType.Long.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.FLOAT =>
+        builder.setFloat(proto.DataType.Float.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.DOUBLE =>
+        builder.setDouble(proto.DataType.Double.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.DATE =>
+        builder.setDate(proto.DataType.Date.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.TIMESTAMP =>
+        builder.setTimestamp(proto.DataType.Timestamp.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.TIMESTAMP_NTZ =>
+        builder.setTimestampNtz(proto.DataType.TimestampNTZ.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.CALENDAR_INTERVAL =>
+        builder.setCalendarInterval(proto.DataType.CalendarInterval.newBuilder.build())
+      case proto.Expression.Literal.LiteralTypeCase.STRUCT =>
+        // The type of the fields will be inferred from the literals of the fields in the struct.
+        builder.setStruct(literal.getStruct.getStructType.getStruct)
+      case _ =>
+        // Not all data types support inferring the data type from the literal at the moment.
+        // e.g. the type of DayTimeInterval contains extra information like start_field and
+        // end_field and cannot be inferred from the literal.
+        return None
+    }
+    Some(builder.build())
+  }
+
+  private def getInferredDataTypeOrThrow(literal: proto.Expression.Literal): proto.DataType = {
+    getInferredDataType(literal).getOrElse {
+      throw InvalidPlanInput(
+        s"Unsupported Literal type for data type inference: ${literal.getLiteralTypeCase}")
     }
   }
 
@@ -386,7 +452,9 @@ object LiteralValueProtoConverter {
     makeMapData(getConverter(map.getKeyType), getConverter(map.getValueType))
   }
 
-  def toCatalystStruct(struct: proto.Expression.Literal.Struct): Any = {
+  def toCatalystStruct(
+      struct: proto.Expression.Literal.Struct,
+      structTypeOpt: Option[proto.DataType.Struct] = None): (Any, proto.DataType.Struct) = {
     def toTuple[A <: Object](data: Seq[A]): Product = {
       try {
         val tupleClass = SparkClassUtils.classForName(s"scala.Tuple${data.length}")
@@ -397,16 +465,78 @@ object LiteralValueProtoConverter {
       }
     }
 
-    val elements = struct.getElementsList.asScala
-    val dataTypes = struct.getStructType.getStruct.getFieldsList.asScala.map(_.getDataType)
-    val structData = elements
-      .zip(dataTypes)
-      .map { case (element, dataType) =>
-        getConverter(dataType)(element)
+    if (struct.hasDataTypeStruct) {
+      // The new way to define and convert structs.
+      val (structData, structType) = if (structTypeOpt.isDefined) {
+        val structFields = structTypeOpt.get.getFieldsList.asScala
+        val structData =
+          struct.getElementsList.asScala.zip(structFields).map { case (element, structField) =>
+            getConverter(structField.getDataType)(element)
+          }
+        (structData, structTypeOpt.get)
+      } else {
+        def protoStructField(
+            name: String,
+            dataType: proto.DataType,
+            nullable: Boolean,
+            metadata: Option[String]): proto.DataType.StructField = {
+          val builder = proto.DataType.StructField
+            .newBuilder()
+            .setName(name)
+            .setDataType(dataType)
+            .setNullable(nullable)
+          metadata.foreach(builder.setMetadata)
+          builder.build()
+        }
+
+        val dataTypeFields = struct.getDataTypeStruct.getFieldsList.asScala
+
+        val structDataAndFields = struct.getElementsList.asScala.zip(dataTypeFields).map {
+          case (element, dataTypeField) =>
+            if (dataTypeField.hasDataType) {
+              (getConverter(dataTypeField.getDataType)(element), dataTypeField)
+            } else {
+              val outerDataType = getInferredDataTypeOrThrow(element)
+              val (value, dataType) =
+                getConverter(outerDataType, inferDataType = true)(element) match {
+                  case LiteralValueWithDataType(value, dataType) => (value, dataType)
+                  case value => (value, outerDataType)
+                }
+              (
+                value,
+                protoStructField(
+                  dataTypeField.getName,
+                  dataType,
+                  dataTypeField.getNullable,
+                  if (dataTypeField.hasMetadata) Some(dataTypeField.getMetadata) else None))
+            }
+        }
+
+        val structType = proto.DataType.Struct
+          .newBuilder()
+          .addAllFields(structDataAndFields.map(_._2).asJava)
+          .build()
+
+        (structDataAndFields.map(_._1), structType)
       }
-      .asInstanceOf[scala.collection.Seq[Object]]
-      .toSeq
+      (toTuple(structData.toSeq.asInstanceOf[Seq[Object]]), structType)
+    } else if (struct.hasStructType) {
+      // For backward compatibility, we still support the old way to define and convert structs.
+      val elements = struct.getElementsList.asScala
+      val dataTypes = struct.getStructType.getStruct.getFieldsList.asScala.map(_.getDataType)
+      val structData = elements
+        .zip(dataTypes)
+        .map { case (element, dataType) =>
+          getConverter(dataType)(element)
+        }
+        .asInstanceOf[scala.collection.Seq[Object]]
+        .toSeq
 
-    toTuple(structData)
+      (toTuple(structData), struct.getStructType.getStruct)
+    } else {
+      throw InvalidPlanInput("Data type information is missing in the struct literal.")
+    }
   }
+
+  private case class LiteralValueWithDataType(value: Any, dataType: proto.DataType)
 }