[SPARK-45896][SQL] Construct ValidateExternalType with the correct expected type

[bersprockets]

What changes were proposed in this pull request?

When creating a serializer for a Map or Seq with an element of type Option, pass an expected type of Option to ValidateExternalType rather than the Option's type argument.

Why are the changes needed?

In 3.4.1, 3.5.0, and master, the following code gets an error:

scala> val df = Seq(Seq(Some(Seq(0)))).toDF("a")
val df = Seq(Seq(Some(Seq(0)))).toDF("a")
org.apache.spark.SparkRuntimeException: [EXPRESSION_ENCODING_FAILED] Failed to encode a value of the expressions: mapobjects(lambdavariable(MapObject, ObjectType(class java.lang.Object), true, -1), mapobjects(lambdavariable(MapObject, ObjectType(class java.lang.Object), true, -2), assertnotnull(validateexternaltype(lambdavariable(MapObject, ObjectType(class java.lang.Object), true, -2), IntegerType, IntegerType)), unwrapoption(ObjectType(interface scala.collection.immutable.Seq), validateexternaltype(lambdavariable(MapObject, ObjectType(class java.lang.Object), true, -1), ArrayType(IntegerType,false), ObjectType(class scala.Option))), None), input[0, scala.collection.immutable.Seq, true], None) AS value#0 to a row. SQLSTATE: 42846
...
Caused by: java.lang.RuntimeException: scala.Some is not a valid external type for schema of array<int>
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.MapObjects_0$(Unknown Source)
...

However, this code works in 3.3.3.

Similarly, this code gets an error:

scala> val df = Seq(Seq(Some(java.sql.Timestamp.valueOf("2023-01-01 00:00:00")))).toDF("a")
val df = Seq(Seq(Some(java.sql.Timestamp.valueOf("2023-01-01 00:00:00")))).toDF("a")
org.apache.spark.SparkRuntimeException: [EXPRESSION_ENCODING_FAILED] Failed to encode a value of the expressions: mapobjects(lambdavariable(MapObject, ObjectType(class java.lang.Object), true, -1), staticinvoke(class org.apache.spark.sql.catalyst.util.DateTimeUtils$, TimestampType, fromJavaTimestamp, unwrapoption(ObjectType(class java.sql.Timestamp), validateexternaltype(lambdavariable(MapObject, ObjectType(class java.lang.Object), true, -1), TimestampType, ObjectType(class scala.Option))), true, false, true), input[0, scala.collection.immutable.Seq, true], None) AS value#0 to a row. SQLSTATE: 42846
...
Caused by: java.lang.RuntimeException: scala.Some is not a valid external type for schema of timestamp
...

As with the first example, this code works in 3.3.3.

SerializerBuildHelper#validateAndSerializeElement will construct ValidateExternalType with an expected type of the Option's type parameter. Therefore, for element types Option[Seq/Date/Timestamp/BigDecimal], ValidateExternalType will try to validate that the element is of the contained type (e.g., BigDecimal) rather than of type Option. Since the element type is of type Option, the validation fails.

Validation currently works by accident for element types Option[Map/<primitive-type], simply because in that case ValidateExternalType ignores that passed expected type and tries to validate based on the encoder's clsTag field (which, for the OptionEncoder, will be class Option).

Does this PR introduce any user-facing change?

Other than fixing the bug, no.

How was this patch tested?

New unit tests.

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

No.

[dongjoon-hyun]

+1, LGTM. Thank you, @bersprockets .

Since this is a regression at Apache Spark 3.4.x, I'd love to have this at Apache Spark 3.4.2 as a release manager.

[dongjoon-hyun]

Merged to master/3.5.

Could you make a backport for branch-3.4? There is a conflict on branch-3.4, @bersprockets .

[bersprockets]

Could you make a backport for branch-3.4? There is a conflict on branch-3.4, @bersprockets .

@dongjoon-hyun I will work on it shortly.

[dongjoon-hyun]

Thank you so much, @bersprockets !