use expressionencoder

Author: Ngone51

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

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql.catalyst
 
-import java.lang.reflect.Modifier
-
 import org.apache.commons.lang3.reflect.ConstructorUtils
 
 import org.apache.spark.internal.Logging
@@ -580,18 +578,14 @@ object ScalaReflection extends ScalaReflection {
   /**
    * Returns the parameter names and types for the primary constructor of this class.
    *
-   * Note that it only works for scala classes with primary constructor.
+   * Note that it only works for scala classes with primary constructor, and currently doesn't
+   * support inner class.
    */
-  def getConstructorParameters(cls: Class[_]): Seq[Class[_]] = {
+  def getConstructorParameters(cls: Class[_]): Seq[(String, Type)] = {
     val m = runtimeMirror(cls.getClassLoader)
     val classSymbol = m.staticClass(cls.getName)
     val t = classSymbol.selfType
-    val dropHead = if (cls.isMemberClass && !Modifier.isStatic(cls.getModifiers)) {
-      1
-    } else {
-      0
-    }
-    getConstructorParameters(t).drop(dropHead).map { case (_, tpe) => getClassFromType(tpe) }
+    getConstructorParameters(t)
   }
 
   /**
@@ -617,15 +611,6 @@ object ScalaReflection extends ScalaReflection {
     }
   }
 
-  def getClassForCaseClass[T: TypeTag]: Option[Class[_]] = {
-    val tpe = localTypeOf[T]
-    if (isSubtype(tpe.dealias, localTypeOf[Product])) {
-      Some(getClassFromType(tpe))
-    } else {
-      None
-    }
-  }
-
   /*
    * Retrieves the runtime class corresponding to the provided type.
    */

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

@@ -256,9 +256,6 @@ class Analyzer(
     Batch("Nondeterministic", Once,
       PullOutNondeterministic),
     Batch("UDF", Once,
-      ResolveCaseClassForUDF,
-      // `ResolveCaseClassForUDF` may generates `NewInstance` so we need to resolve it
-      ResolveNewInstance,
       HandleNullInputsForUDF),
     Batch("UpdateNullability", Once,
       UpdateAttributeNullability),
@@ -2744,39 +2741,6 @@ class Analyzer(
     }
   }
 
-  object ResolveCaseClassForUDF extends Rule[LogicalPlan] {
-    override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
-      case p if !p.resolved => p // Skip unresolved nodes.
-
-      case p => p transformExpressionsUp {
-
-        case udf @ ScalaUDF(_, _, inputs, _, inputCaseClass, _, _, _, _) =>
-          if (inputCaseClass.exists(_.isDefined)) {
-            assert(inputs.size == inputCaseClass.size)
-            val newInputs = inputs.zip(inputCaseClass).map {
-              case (input, clazzOpt) =>
-                if (clazzOpt.isDefined) {
-                  val clazz = clazzOpt.get
-                  assert(input.dataType.isInstanceOf[StructType],
-                    s"expects StructType, but got ${input.dataType}")
-                  val dataType = input.dataType.asInstanceOf[StructType]
-                  val args = dataType.toAttributes.zipWithIndex.map { case (a, i) =>
-                    GetStructField(input, i, Some(a.name))
-                  }
-                  NewInstance(clazz, args, ObjectType(clazz))
-                } else {
-                  input
-                }
-            }
-            // assign Nil inputCaseClass to avoid applying this rule for multiple times
-            udf.copy(children = newInputs, inputCaseClass = Nil)
-          } else {
-            udf
-          }
-      }
-    }
-  }
-
   /**
    * Check and add proper window frames for all window functions.
    */