improve

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

@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
-import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, DataType}
+import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, DataType, DecimalType, StructType}
 
 /**
  * User-defined function.
@@ -112,17 +112,21 @@ case class ScalaUDF(
       val encoder = inputEncoders(i)
       encoder match {
         case Some(enc) =>
-          val fromRow = enc.resolveAndBind().createDeserializer()
           if (enc.isSerializedAsStructForTopLevel) {
+            val fromRow = enc.resolveAndBind().createDeserializer()
             row: Any => fromRow(row.asInstanceOf[InternalRow])
           } else {
+            val child = children(i)
+            val attrs = new StructType().add(s"$child", child.dataType).toAttributes
+            val fromRow = enc.resolveAndBind(attrs).createDeserializer()
+
             value: Any =>
               val row = new GenericInternalRow(1)
               row.update(0, value)
               fromRow(row)
           }
 
-        case None => createToScalaConverter(dataType)
+        case _ => createToScalaConverter(dataType)
       }
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/UDFSuite.scala

@@ -608,6 +608,13 @@ class UDFSuite extends QueryTest with SharedSparkSession {
     val f3 = (s: Map[TestData, TestData]) => s.keys.head.key * s.values.head.value.toInt
     val myUdf3 = udf(f3)
     val df3 = Seq(("data", Map(TestData(50, "2") -> TestData(50, "2")))).toDF("col1", "col2")
-    checkAnswer(df2.select(myUdf2(Column("col2"))), Row(100) :: Nil)
+    checkAnswer(df3.select(myUdf3(Column("col2"))), Row(100) :: Nil)
+  }
+
+  test("case class as element of tuple") {
+    val f = (s: (TestData, Int)) => s._1.key * s._2
+    val myUdf = udf(f)
+    val df = Seq(("data", (TestData(50, "2"), 2))).toDF("col1", "col2")
+    checkAnswer(df.select(myUdf(Column("col2"))), Row(100) :: Nil)
   }
 }