fix

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -2858,9 +2858,10 @@ class Analyzer(
           if encoders.nonEmpty && desers.isEmpty =>
           val deserializers = encoders.zipWithIndex.map { case (encOpt, i) =>
             val dataType = inputs(i).dataType
-            if (CatalystTypeConverters.isPrimitive(dataType)) {
-              // primitive data types do not rely on `ExpressionEncoder` to
-              // convert data, see `ScalaUDF.scalaConverter`.
+            if (CatalystTypeConverters.isPrimitive(dataType) ||
+              dataType.isInstanceOf[UserDefinedType[_]]) {
+              // primitive/UDT data types use `CatalystTypeConverters` to
+              // convert internal data to external data.
               None
             } else {
               encOpt.map { enc =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUDF.scala

@@ -58,6 +58,13 @@ case class ScalaUDF(
 
   override lazy val deterministic: Boolean = udfDeterministic && children.forall(_.deterministic)
 
+  override lazy val canonicalized: Expression = {
+    val canonicalizedChildren = children.map(_.canonicalized)
+    // if the canonicalized children and inputEncoders are equal,
+    // then we must have equal inputDeserializers as well.
+    this.copy(inputDeserializers = Nil).withNewChildren(canonicalizedChildren)
+  }
+
   override def toString: String = s"${udfName.getOrElse("UDF")}(${children.mkString(", ")})"
 
   /**
@@ -127,6 +134,7 @@ case class ScalaUDF(
               fromRow(row)
           }
 
+          // e.g. for UDF types
         case _ => createToScalaConverter(dataType)
       }
     }