[SPARK-23593][SQL] Add interpreted execution for InitializeJavaBean expression

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

@@ -1261,8 +1261,31 @@ case class InitializeJavaBean(beanInstance: Expression, setters: Map[String, Exp
   override def children: Seq[Expression] = beanInstance +: setters.values.toSeq
   override def dataType: DataType = beanInstance.dataType
 
-  override def eval(input: InternalRow): Any =
-    throw new UnsupportedOperationException("Only code-generated evaluation is supported.")
+  private lazy val resolvedSetters = {
+    val ObjectType(beanClass) = beanInstance.dataType
+
+    setters.map { case (setterMethod, fieldExpr) =>
+      val foundMethods = beanClass.getMethods.filter { method =>
+        method.getName == setterMethod && Modifier.isPublic(method.getModifiers) &&
+          method.getParameterTypes.length == 1
+      }
+      assert(foundMethods.length == 1,
+        throw new RuntimeException("The Java Bean class should have only one " +
+          s"setter $setterMethod method, but ${foundMethods.length} methods found."))
+      (foundMethods.head, fieldExpr)
+    }
+  }
+
+  override def eval(input: InternalRow): Any = {
+    val instance = beanInstance.eval(input).asInstanceOf[Object]
+    if (instance != null) {
+      resolvedSetters.foreach { case (setterMethod, fieldExpr) =>
+        val fieldValue = fieldExpr.eval(input).asInstanceOf[Object]
+        setterMethod.invoke(instance, fieldValue)
+      }
+    }
+    instance
+  }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val instanceGen = beanInstance.genCode(ctx)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionEvalHelper.scala

@@ -49,7 +49,8 @@ trait ExpressionEvalHelper extends GeneratorDrivenPropertyChecks {
       expression: => Expression, expected: Any, inputRow: InternalRow = EmptyRow): Unit = {
     val serializer = new JavaSerializer(new SparkConf()).newInstance
     val resolver = ResolveTimeZone(new SQLConf)
-    val expr = resolver.resolveTimeZones(serializer.deserialize(serializer.serialize(expression)))
+    // Make it as method to obtain fresh expression everytime.
+    def expr = resolver.resolveTimeZones(serializer.deserialize(serializer.serialize(expression)))
     val catalystValue = CatalystTypeConverters.convertToCatalyst(expected)
     checkEvaluationWithoutCodegen(expr, catalystValue, inputRow)
     checkEvaluationWithGeneratedMutableProjection(expr, catalystValue, inputRow)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ObjectExpressionsSuite.scala

@@ -68,6 +68,23 @@ class ObjectExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
       mapEncoder.serializer.head, mapExpected, mapInputRow)
   }
 
+  test("SPARK-23593: InitializeJavaBean should support interpreted execution") {
+    val list = new java.util.LinkedList[Int]()
+    list.add(1)
+
+    val initializeBean = InitializeJavaBean(Literal.fromObject(new java.util.LinkedList[Int]),
+      Map("add" -> Literal(1)))
+    checkEvaluation(initializeBean, list, InternalRow.fromSeq(Seq()))
+
+    val errMsg = intercept[RuntimeException] {
+      val initializeWithNonexistingMethod = InitializeJavaBean(
+        Literal.fromObject(new java.util.LinkedList[Int]),
+        Map("nonexisting" -> Literal(1)))
+      evaluate(initializeWithNonexistingMethod, InternalRow.fromSeq(Seq()))
+    }.getMessage
+    assert(errMsg.contains("but 0 methods found."))
+  }
+
   test("SPARK-23585: UnwrapOption should support interpreted execution") {
     val cls = classOf[Option[Int]]
     val inputObject = BoundReference(0, ObjectType(cls), nullable = true)