[SPARK-41993][SQL] Move RowEncoder to AgnosticEncoders

### What changes were proposed in this pull request?
This PR makes `RowEncoder` produce an `AgnosticEncoder`. The expression generation for these encoders is moved to `ScalaReflection` (this will be moved out in a subsequent PR).

The generated serializer and deserializer expressions will slightly change for both schema and type based encoders. These are not semantically different from the old expressions. Concretely the following changes have been introduced:
- There is more type validation in maps/arrays/seqs for type based encoders. This should be a positive change, since it disallows users to pass wrong data through erasure hackd.
- Array/Seq serialization is a bit more strict. In the old scenario it was possible to pass in sequences/arrays with the wrong type and/or nullability.

### Why are the changes needed?
For the Spark Connect Scala Client we also want to be able to use `Row` based results.

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

### How was this patch tested?
This is a refactoring, existing tests should be sufficient.

Closes #39517 from hvanhovell/SPARK-41993.

Authored-by: Herman van Hovell <herman@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: hvanhovell

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala

@@ -423,10 +423,10 @@ object JavaTypeInference {
         case c if c == classOf[java.time.Period] => createSerializerForJavaPeriod(inputObject)
 
         case c if c == classOf[java.math.BigInteger] =>
-          createSerializerForJavaBigInteger(inputObject)
+          createSerializerForBigInteger(inputObject)
 
         case c if c == classOf[java.math.BigDecimal] =>
-          createSerializerForJavaBigDecimal(inputObject)
+          createSerializerForBigDecimal(inputObject)
 
         case c if c == classOf[java.lang.Boolean] => createSerializerForBoolean(inputObject)
         case c if c == classOf[java.lang.Byte] => createSerializerForByte(inputObject)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -158,20 +158,29 @@ object SerializerBuildHelper {
       returnNullable = false)
   }
 
-  def createSerializerForJavaBigDecimal(inputObject: Expression): Expression = {
+  def createSerializerForBigDecimal(inputObject: Expression): Expression = {
+    createSerializerForBigDecimal(inputObject, DecimalType.SYSTEM_DEFAULT)
+  }
+
+  def createSerializerForBigDecimal(inputObject: Expression, dt: DecimalType): Expression = {
     CheckOverflow(StaticInvoke(
       Decimal.getClass,
-      DecimalType.SYSTEM_DEFAULT,
+      dt,
       "apply",
       inputObject :: Nil,
-      returnNullable = false), DecimalType.SYSTEM_DEFAULT, nullOnOverflow)
+      returnNullable = false), dt, nullOnOverflow)
   }
 
-  def createSerializerForScalaBigDecimal(inputObject: Expression): Expression = {
-    createSerializerForJavaBigDecimal(inputObject)
+  def createSerializerForAnyDecimal(inputObject: Expression, dt: DecimalType): Expression = {
+    CheckOverflow(StaticInvoke(
+      Decimal.getClass,
+      dt,
+      "fromDecimal",
+      inputObject :: Nil,
+      returnNullable = false), dt, nullOnOverflow)
   }
 
-  def createSerializerForJavaBigInteger(inputObject: Expression): Expression = {
+  def createSerializerForBigInteger(inputObject: Expression): Expression = {
     CheckOverflow(StaticInvoke(
       Decimal.getClass,
       DecimalType.BigIntDecimal,
@@ -180,10 +189,6 @@ object SerializerBuildHelper {
       returnNullable = false), DecimalType.BigIntDecimal, nullOnOverflow)
   }
 
-  def createSerializerForScalaBigInt(inputObject: Expression): Expression = {
-    createSerializerForJavaBigInteger(inputObject)
-  }
-
   def createSerializerForPrimitiveArray(
       inputObject: Expression,
       dataType: DataType): Expression = {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala

@@ -16,28 +16,33 @@
  */
 package org.apache.spark.sql.catalyst.encoders
 
+import java.{sql => jsql}
 import java.math.{BigDecimal => JBigDecimal, BigInteger => JBigInt}
 import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period}
 
 import scala.reflect.{classTag, ClassTag}
 
-import org.apache.spark.sql.Encoder
+import org.apache.spark.sql.{Encoder, Row}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.CalendarInterval
 
 /**
  * A non implementation specific encoder. This encoder containers all the information needed
  * to generate an implementation specific encoder (e.g. InternalRow <=> Custom Object).
+ *
+ * The input of the serialization does not need to match the external type of the encoder. This is
+ * called lenient serialization. An example of this is lenient date serialization, in this case both
+ * [[java.sql.Date]] and [[java.time.LocalDate]] are allowed. Deserialization is never lenient; it
+ * will always produce instance of the external type.
  */
 trait AgnosticEncoder[T] extends Encoder[T] {
   def isPrimitive: Boolean
   def nullable: Boolean = !isPrimitive
   def dataType: DataType
   override def schema: StructType = StructType(StructField("value", dataType, nullable) :: Nil)
+  def lenientSerialization: Boolean = false
 }
 
-// TODO check RowEncoder
-// TODO check BeanEncoder
 object AgnosticEncoders {
   case class OptionEncoder[E](elementEncoder: AgnosticEncoder[E])
     extends AgnosticEncoder[Option[E]] {
@@ -46,35 +51,48 @@ object AgnosticEncoders {
     override val clsTag: ClassTag[Option[E]] = ClassTag(classOf[Option[E]])
   }
 
-  case class ArrayEncoder[E](element: AgnosticEncoder[E])
+  case class ArrayEncoder[E](element: AgnosticEncoder[E], containsNull: Boolean)
     extends AgnosticEncoder[Array[E]] {
     override def isPrimitive: Boolean = false
-    override def dataType: DataType = ArrayType(element.dataType, element.nullable)
+    override def dataType: DataType = ArrayType(element.dataType, containsNull)
     override val clsTag: ClassTag[Array[E]] = element.clsTag.wrap
   }
 
-  case class IterableEncoder[C <: Iterable[E], E](
+  /**
+   * Encoder for collections.
+   *
+   * This encoder can be lenient for [[Row]] encoders. In that case we allow [[Seq]], primitive
+   * array (if any), and generic arrays as input.
+   */
+  case class IterableEncoder[C, E](
       override val clsTag: ClassTag[C],
-      element: AgnosticEncoder[E])
+      element: AgnosticEncoder[E],
+      containsNull: Boolean,
+      override val lenientSerialization: Boolean)
     extends AgnosticEncoder[C] {
     override def isPrimitive: Boolean = false
-    override val dataType: DataType = ArrayType(element.dataType, element.nullable)
+    override val dataType: DataType = ArrayType(element.dataType, containsNull)
   }
 
   case class MapEncoder[C, K, V](
       override val clsTag: ClassTag[C],
       keyEncoder: AgnosticEncoder[K],
-      valueEncoder: AgnosticEncoder[V])
+      valueEncoder: AgnosticEncoder[V],
+      valueContainsNull: Boolean)
     extends AgnosticEncoder[C] {
     override def isPrimitive: Boolean = false
     override val dataType: DataType = MapType(
       keyEncoder.dataType,
       valueEncoder.dataType,
-      valueEncoder.nullable)
+      valueContainsNull)
   }
 
-  case class EncoderField(name: String, enc: AgnosticEncoder[_]) {
-    def structField: StructField = StructField(name, enc.dataType, enc.nullable)
+  case class EncoderField(
+      name: String,
+      enc: AgnosticEncoder[_],
+      nullable: Boolean,
+      metadata: Metadata) {
+    def structField: StructField = StructField(name, enc.dataType, nullable, metadata)
   }
 
   // This supports both Product and DefinedByConstructorParams
@@ -87,6 +105,13 @@ object AgnosticEncoders {
     override def dataType: DataType = schema
   }
 
+  case class RowEncoder(fields: Seq[EncoderField]) extends AgnosticEncoder[Row] {
+    override def isPrimitive: Boolean = false
+    override val schema: StructType = StructType(fields.map(_.structField))
+    override def dataType: DataType = schema
+    override def clsTag: ClassTag[Row] = classTag[Row]
+  }
+
   // This will only work for encoding from/to Sparks' InternalRow format.
   // It is here for compatibility.
   case class UDTEncoder[E >: Null](
@@ -116,39 +141,74 @@ object AgnosticEncoders {
   }
 
   // Primitive encoders
-  case object PrimitiveBooleanEncoder extends LeafEncoder[Boolean](BooleanType)
-  case object PrimitiveByteEncoder extends LeafEncoder[Byte](ByteType)
-  case object PrimitiveShortEncoder extends LeafEncoder[Short](ShortType)
-  case object PrimitiveIntEncoder extends LeafEncoder[Int](IntegerType)
-  case object PrimitiveLongEncoder extends LeafEncoder[Long](LongType)
-  case object PrimitiveFloatEncoder extends LeafEncoder[Float](FloatType)
-  case object PrimitiveDoubleEncoder extends LeafEncoder[Double](DoubleType)
+  abstract class PrimitiveLeafEncoder[E : ClassTag](dataType: DataType)
+    extends LeafEncoder[E](dataType)
+  case object PrimitiveBooleanEncoder extends PrimitiveLeafEncoder[Boolean](BooleanType)
+  case object PrimitiveByteEncoder extends PrimitiveLeafEncoder[Byte](ByteType)
+  case object PrimitiveShortEncoder extends PrimitiveLeafEncoder[Short](ShortType)
+  case object PrimitiveIntEncoder extends PrimitiveLeafEncoder[Int](IntegerType)
+  case object PrimitiveLongEncoder extends PrimitiveLeafEncoder[Long](LongType)
+  case object PrimitiveFloatEncoder extends PrimitiveLeafEncoder[Float](FloatType)
+  case object PrimitiveDoubleEncoder extends PrimitiveLeafEncoder[Double](DoubleType)
 
   // Primitive wrapper encoders.
-  case object NullEncoder extends LeafEncoder[java.lang.Void](NullType)
-  case object BoxedBooleanEncoder extends LeafEncoder[java.lang.Boolean](BooleanType)
-  case object BoxedByteEncoder extends LeafEncoder[java.lang.Byte](ByteType)
-  case object BoxedShortEncoder extends LeafEncoder[java.lang.Short](ShortType)
-  case object BoxedIntEncoder extends LeafEncoder[java.lang.Integer](IntegerType)
-  case object BoxedLongEncoder extends LeafEncoder[java.lang.Long](LongType)
-  case object BoxedFloatEncoder extends LeafEncoder[java.lang.Float](FloatType)
-  case object BoxedDoubleEncoder extends LeafEncoder[java.lang.Double](DoubleType)
+  abstract class BoxedLeafEncoder[E : ClassTag, P](
+      dataType: DataType,
+      val primitive: PrimitiveLeafEncoder[P])
+    extends LeafEncoder[E](dataType)
+  case object BoxedBooleanEncoder
+    extends BoxedLeafEncoder[java.lang.Boolean, Boolean](BooleanType, PrimitiveBooleanEncoder)
+  case object BoxedByteEncoder
+    extends BoxedLeafEncoder[java.lang.Byte, Byte](ByteType, PrimitiveByteEncoder)
+  case object BoxedShortEncoder
+    extends BoxedLeafEncoder[java.lang.Short, Short](ShortType, PrimitiveShortEncoder)
+  case object BoxedIntEncoder
+    extends BoxedLeafEncoder[java.lang.Integer, Int](IntegerType, PrimitiveIntEncoder)
+  case object BoxedLongEncoder
+    extends BoxedLeafEncoder[java.lang.Long, Long](LongType, PrimitiveLongEncoder)
+  case object BoxedFloatEncoder
+    extends BoxedLeafEncoder[java.lang.Float, Float](FloatType, PrimitiveFloatEncoder)
+  case object BoxedDoubleEncoder
+    extends BoxedLeafEncoder[java.lang.Double, Double](DoubleType, PrimitiveDoubleEncoder)
 
   // Nullable leaf encoders
+  case object NullEncoder extends LeafEncoder[java.lang.Void](NullType)
   case object StringEncoder extends LeafEncoder[String](StringType)
   case object BinaryEncoder extends LeafEncoder[Array[Byte]](BinaryType)
-  case object SparkDecimalEncoder extends LeafEncoder[Decimal](DecimalType.SYSTEM_DEFAULT)
-  case object ScalaDecimalEncoder extends LeafEncoder[BigDecimal](DecimalType.SYSTEM_DEFAULT)
-  case object JavaDecimalEncoder extends LeafEncoder[JBigDecimal](DecimalType.SYSTEM_DEFAULT)
   case object ScalaBigIntEncoder extends LeafEncoder[BigInt](DecimalType.BigIntDecimal)
   case object JavaBigIntEncoder extends LeafEncoder[JBigInt](DecimalType.BigIntDecimal)
   case object CalendarIntervalEncoder extends LeafEncoder[CalendarInterval](CalendarIntervalType)
   case object DayTimeIntervalEncoder extends LeafEncoder[Duration](DayTimeIntervalType())
   case object YearMonthIntervalEncoder extends LeafEncoder[Period](YearMonthIntervalType())
-  case object DateEncoder extends LeafEncoder[java.sql.Date](DateType)
-  case object LocalDateEncoder extends LeafEncoder[LocalDate](DateType)
-  case object TimestampEncoder extends LeafEncoder[java.sql.Timestamp](TimestampType)
-  case object InstantEncoder extends LeafEncoder[Instant](TimestampType)
+  case class DateEncoder(override val lenientSerialization: Boolean)
+    extends LeafEncoder[jsql.Date](DateType)
+  case class LocalDateEncoder(override val lenientSerialization: Boolean)
+    extends LeafEncoder[LocalDate](DateType)
+  case class TimestampEncoder(override val lenientSerialization: Boolean)
+    extends LeafEncoder[jsql.Timestamp](TimestampType)
+  case class InstantEncoder(override val lenientSerialization: Boolean)
+    extends LeafEncoder[Instant](TimestampType)
   case object LocalDateTimeEncoder extends LeafEncoder[LocalDateTime](TimestampNTZType)
+
+  case class SparkDecimalEncoder(dt: DecimalType) extends LeafEncoder[Decimal](dt)
+  case class ScalaDecimalEncoder(dt: DecimalType) extends LeafEncoder[BigDecimal](dt)
+  case class JavaDecimalEncoder(dt: DecimalType, override val lenientSerialization: Boolean)
+    extends LeafEncoder[JBigDecimal](dt)
+
+  val STRICT_DATE_ENCODER: DateEncoder = DateEncoder(lenientSerialization = false)
+  val STRICT_LOCAL_DATE_ENCODER: LocalDateEncoder = LocalDateEncoder(lenientSerialization = false)
+  val STRICT_TIMESTAMP_ENCODER: TimestampEncoder = TimestampEncoder(lenientSerialization = false)
+  val STRICT_INSTANT_ENCODER: InstantEncoder = InstantEncoder(lenientSerialization = false)
+  val LENIENT_DATE_ENCODER: DateEncoder = DateEncoder(lenientSerialization = true)
+  val LENIENT_LOCAL_DATE_ENCODER: LocalDateEncoder = LocalDateEncoder(lenientSerialization = true)
+  val LENIENT_TIMESTAMP_ENCODER: TimestampEncoder = TimestampEncoder(lenientSerialization = true)
+  val LENIENT_INSTANT_ENCODER: InstantEncoder = InstantEncoder(lenientSerialization = true)
+
+  val DEFAULT_SPARK_DECIMAL_ENCODER: SparkDecimalEncoder =
+    SparkDecimalEncoder(DecimalType.SYSTEM_DEFAULT)
+  val DEFAULT_SCALA_DECIMAL_ENCODER: ScalaDecimalEncoder =
+    ScalaDecimalEncoder(DecimalType.SYSTEM_DEFAULT)
+  val DEFAULT_JAVA_DECIMAL_ENCODER: JavaDecimalEncoder =
+    JavaDecimalEncoder(DecimalType.SYSTEM_DEFAULT, lenientSerialization = false)
 }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/AgnosticEncoder.scala

@@ -47,7 +47,10 @@ import org.apache.spark.util.Utils
 object ExpressionEncoder {
 
   def apply[T : TypeTag](): ExpressionEncoder[T] = {
-    val enc = ScalaReflection.encoderFor[T]
+    apply(ScalaReflection.encoderFor[T])
+  }
+
+  def apply[T](enc: AgnosticEncoder[T]): ExpressionEncoder[T] = {
     new ExpressionEncoder[T](
       ScalaReflection.serializerFor(enc),
       ScalaReflection.deserializerFor(enc),