[SPARK-52930][CONNECT] Use DataType.Array/Map for Array/Map Literals

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

This PR introduces a transition to use `DataType.Array` and `DataType.Map` for array and map literals throughout the Spark Connect codebase for Array/Map Literals.

While the Spark Connect server supports both new and old data type fields, in this change, the new data type fields are set only in ColumnNodeToProtoConverter for the Spark Connect Scala client. All other components (e.g., ML, Python) still use the old data type fields because literal values are used not only in requests but also in responses, making it difficult to maintain compatibility—clients using older versions may not recognize the new fields in the response. Deprecation and the transition to the new fields require a gradual migration. The key changes include:

**Protocol Buffer Updates:**
- Modified `expressions.proto` to add new `data_type` fields for `Array` and `Map` messages
- Deprecated existing `element_type`, `key_type`, and `value_type` fields in favor of the unified `data_type` approach
- Updated generated protocol buffer files (`expressions_pb2.py`, `expressions_pb2.pyi`) to reflect these changes

**Core Implementation Changes:**
- Enhanced `LiteralValueProtoConverter.scala` with new internal method `toLiteralProtoBuilderInternal` that accepts `ToLiteralProtoOptions`
- Updated `LiteralExpressionProtoConverter.scala` to support inference of array and map data types
- Modified `columnNodeSupport.scala` to use the new `toLiteralProtoBuilderWithOptions` method with `useDeprecatedDataTypeFields` set to `false`

### Why are the changes needed?

The changes are needed to improve Spark's data type handling for array and map literals:

- **Nullability of Array/Map literals are now included in the DataType.Array/Map**: This ensures that nullability information is properly captured and handled within the data type structure itself.

- **Work better with type inference by including all type information in one field**: By consolidating all type information into a single field, it is easier to infer data types for complex data structures.

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

Yes. Previously, the nullability of arrays and map values using typedlit was not preserved (which I believe was a bug). It is now preserved. Please see the changes in ClientE2ETestSuite for details.

### How was this patch tested?

`build/sbt "connect/testOnly *LiteralExpressionProtoConverterSuite"`
`build/sbt "connect-client-jvm/testOnly *ClientE2ETestSuite -- -z SPARK-52930"`

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

Generated-by: Cursor 1.4.5

Closes #51653 from heyihong/SPARK-52930.

Authored-by: Yihong He <heyihong.cn@gmail.com>
Signed-off-by: Ruifeng Zheng <ruifengz@apache.org>

Author: heyihong

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

@@ -474,27 +474,51 @@ class Expression(google.protobuf.message.Message):
 
             ELEMENT_TYPE_FIELD_NUMBER: builtins.int
             ELEMENTS_FIELD_NUMBER: builtins.int
+            DATA_TYPE_FIELD_NUMBER: builtins.int
             @property
-            def element_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType: ...
+            def element_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType:
+                """(Deprecated) The element type of the array.
+
+                This field is deprecated since Spark 4.1+ and should only be set
+                if the data_type field is not set. Use data_type field instead.
+                """
             @property
             def elements(
                 self,
             ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
                 global___Expression.Literal
-            ]: ...
+            ]:
+                """The literal values that make up the array elements."""
+            @property
+            def data_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType.Array:
+                """The type of the array.
+
+                If the element type can be inferred from the first element of the elements field,
+                then you don't need to set data_type.element_type to save space. On the other hand,
+                redundant type information is also acceptable.
+                """
             def __init__(
                 self,
                 *,
                 element_type: pyspark.sql.connect.proto.types_pb2.DataType | None = ...,
                 elements: collections.abc.Iterable[global___Expression.Literal] | None = ...,
+                data_type: pyspark.sql.connect.proto.types_pb2.DataType.Array | None = ...,
             ) -> None: ...
             def HasField(
-                self, field_name: typing_extensions.Literal["element_type", b"element_type"]
+                self,
+                field_name: typing_extensions.Literal[
+                    "data_type", b"data_type", "element_type", b"element_type"
+                ],
             ) -> builtins.bool: ...
             def ClearField(
                 self,
                 field_name: typing_extensions.Literal[
-                    "element_type", b"element_type", "elements", b"elements"
+                    "data_type",
+                    b"data_type",
+                    "element_type",
+                    b"element_type",
+                    "elements",
+                    b"elements",
                 ],
             ) -> None: ...
 
@@ -505,39 +529,63 @@ class Expression(google.protobuf.message.Message):
             VALUE_TYPE_FIELD_NUMBER: builtins.int
             KEYS_FIELD_NUMBER: builtins.int
             VALUES_FIELD_NUMBER: builtins.int
+            DATA_TYPE_FIELD_NUMBER: builtins.int
             @property
-            def key_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType: ...
+            def key_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType:
+                """(Deprecated) The key type of the map.
+
+                This field is deprecated since Spark 4.1+ and should only be set
+                if the data_type field is not set. Use data_type field instead.
+                """
             @property
-            def value_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType: ...
+            def value_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType:
+                """(Deprecated) The value type of the map.
+
+                This field is deprecated since Spark 4.1+ and should only be set
+                if the data_type field is not set. Use data_type field instead.
+                """
             @property
             def keys(
                 self,
             ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
                 global___Expression.Literal
-            ]: ...
+            ]:
+                """The literal keys that make up the map."""
             @property
             def values(
                 self,
             ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
                 global___Expression.Literal
-            ]: ...
+            ]:
+                """The literal values that make up the map."""
+            @property
+            def data_type(self) -> pyspark.sql.connect.proto.types_pb2.DataType.Map:
+                """The type of the map.
+
+                If the key/value types can be inferred from the first element of the keys/values fields,
+                then you don't need to set data_type.key_type/data_type.value_type to save space.
+                On the other hand, redundant type information is also acceptable.
+                """
             def __init__(
                 self,
                 *,
                 key_type: pyspark.sql.connect.proto.types_pb2.DataType | None = ...,
                 value_type: pyspark.sql.connect.proto.types_pb2.DataType | None = ...,
                 keys: collections.abc.Iterable[global___Expression.Literal] | None = ...,
                 values: collections.abc.Iterable[global___Expression.Literal] | None = ...,
+                data_type: pyspark.sql.connect.proto.types_pb2.DataType.Map | None = ...,
             ) -> None: ...
             def HasField(
                 self,
                 field_name: typing_extensions.Literal[
-                    "key_type", b"key_type", "value_type", b"value_type"
+                    "data_type", b"data_type", "key_type", b"key_type", "value_type", b"value_type"
                 ],
             ) -> builtins.bool: ...
             def ClearField(
                 self,
                 field_name: typing_extensions.Literal[
+                    "data_type",
+                    b"data_type",
                     "key_type",
                     b"key_type",
                     "keys",

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

@@ -1687,6 +1687,35 @@ class ClientE2ETestSuite
       assert(df.count() == 100)
     }
   }
+
+  test("SPARK-52930: the nullability of arrays should be preserved using typedlit") {
+    val arrays = Seq(
+      (typedlit(Array[Int]()), false),
+      (typedlit(Array[Int](1)), false),
+      (typedlit(Array[Integer]()), true),
+      (typedlit(Array[Integer](1)), true))
+    for ((array, containsNull) <- arrays) {
+      val df = spark.sql("select 1").select(array)
+      df.createOrReplaceTempView("test_array_nullability")
+      val schema = spark.sql("select * from test_array_nullability").schema
+      assert(schema.fields.head.dataType.asInstanceOf[ArrayType].containsNull === containsNull)
+    }
+  }
+
+  test("SPARK-52930: the nullability of map values should be preserved using typedlit") {
+    val maps = Seq(
+      (typedlit(Map[String, Int]()), false),
+      (typedlit(Map[String, Int]("a" -> 1)), false),
+      (typedlit(Map[String, Integer]()), true),
+      (typedlit(Map[String, Integer]("a" -> 1)), true))
+    for ((map, valueContainsNull) <- maps) {
+      val df = spark.sql("select 1").select(map)
+      df.createOrReplaceTempView("test_map_nullability")
+      val schema = spark.sql("select * from test_map_nullability").schema
+      assert(
+        schema.fields.head.dataType.asInstanceOf[MapType].valueContainsNull === valueContainsNull)
+    }
+  }
 }
 
 private[sql] case class ClassData(a: String, b: Int)

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

@@ -215,15 +215,48 @@ message Expression {
     }
 
     message Array {
-      DataType element_type = 1;
+      // (Deprecated) The element type of the array.
+      //
+      // This field is deprecated since Spark 4.1+ and should only be set
+      // if the data_type field is not set. Use data_type field instead.
+      DataType element_type = 1 [deprecated = true];
+
+      // The literal values that make up the array elements.
       repeated Literal elements = 2;
+
+      // The type of the array.
+      //
+      // If the element type can be inferred from the first element of the elements field,
+      // then you don't need to set data_type.element_type to save space. On the other hand,
+      // redundant type information is also acceptable.
+      DataType.Array data_type = 3;
     }
 
     message Map {
-      DataType key_type = 1;
-      DataType value_type = 2;
+      // (Deprecated) The key type of the map.
+      //
+      // This field is deprecated since Spark 4.1+ and should only be set
+      // if the data_type field is not set. Use data_type field instead.
+      DataType key_type = 1 [deprecated = true];
+
+      // (Deprecated) The value type of the map.
+      //
+      // This field is deprecated since Spark 4.1+ and should only be set
+      // if the data_type field is not set. Use data_type field instead.
+      DataType value_type = 2 [deprecated = true];
+
+      // The literal keys that make up the map.
       repeated Literal keys = 3;
+
+      // The literal values that make up the map.
       repeated Literal values = 4;
+
+      // The type of the map.
+      //
+      // If the key/value types can be inferred from the first element of the keys/values fields,
+      // then you don't need to set data_type.key_type/data_type.value_type to save space.
+      // On the other hand, redundant type information is also acceptable.
+      DataType.Map data_type = 5;
     }
 
     message Struct {

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

@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
 import org.apache.spark.sql.connect.ConnectConversions._
 import org.apache.spark.sql.connect.common.DataTypeProtoConverter
-import org.apache.spark.sql.connect.common.LiteralValueProtoConverter.toLiteralProtoBuilder
+import org.apache.spark.sql.connect.common.LiteralValueProtoConverter.{toLiteralProtoBuilderWithOptions, ToLiteralProtoOptions}
 import org.apache.spark.sql.expressions.{Aggregator, UserDefinedAggregateFunction, UserDefinedAggregator, UserDefinedFunction}
 import org.apache.spark.sql.internal.{Alias, CaseWhenOtherwise, Cast, ColumnNode, ColumnNodeLike, InvokeInlineUserDefinedFunction, LambdaFunction, LazyExpression, Literal, SortOrder, SqlExpression, SubqueryExpression, SubqueryType, UnresolvedAttribute, UnresolvedExtractValue, UnresolvedFunction, UnresolvedNamedLambdaVariable, UnresolvedRegex, UnresolvedStar, UpdateFields, Window, WindowFrame}
 
@@ -65,11 +65,12 @@ object ColumnNodeToProtoConverter extends (ColumnNode => proto.Expression) {
     val builder = proto.Expression.newBuilder()
     val n = additionalTransformation.map(_(node)).getOrElse(node)
     n match {
-      case Literal(value, None, _) =>
-        builder.setLiteral(toLiteralProtoBuilder(value))
-
-      case Literal(value, Some(dataType), _) =>
-        builder.setLiteral(toLiteralProtoBuilder(value, dataType))
+      case Literal(value, dataTypeOpt, _) =>
+        builder.setLiteral(
+          toLiteralProtoBuilderWithOptions(
+            value,
+            dataTypeOpt,
+            ToLiteralProtoOptions(useDeprecatedDataTypeFields = false)))
 
       case u @ UnresolvedAttribute(unparsedIdentifier, planId, isMetadataColumn, _) =>
         val escapedName = u.sql