From 4926a7c2f0a47b562f99dbb4f1ca17adb3192061 Mon Sep 17 00:00:00 2001
From: Takeshi Yamamuro <yamamuro@apache.org>
Date: Tue, 24 Apr 2018 17:52:05 +0200
Subject: [PATCH] [SPARK-23589][SQL][FOLLOW-UP] Reuse InternalRow in
 ExternalMapToCatalyst eval

## What changes were proposed in this pull request?
This pr is a follow-up of #20980 and fixes code to reuse `InternalRow` for converting input keys/values in `ExternalMapToCatalyst` eval.

## How was this patch tested?
Existing tests.

Author: Takeshi Yamamuro <yamamuro@apache.org>

Closes #21137 from maropu/SPARK-23589-FOLLOWUP.
---
 .../expressions/objects/objects.scala         | 92 ++++++++++---------
 1 file changed, 50 insertions(+), 42 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala
index 9c7e76467d153..f974fd81fc788 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala
@@ -1255,53 +1255,61 @@ case class ExternalMapToCatalyst private(
   override def dataType: MapType = MapType(
     keyConverter.dataType, valueConverter.dataType, valueContainsNull = valueConverter.nullable)
 
-  private lazy val mapCatalystConverter: Any => (Array[Any], Array[Any]) = child.dataType match {
-    case ObjectType(cls) if classOf[java.util.Map[_, _]].isAssignableFrom(cls) =>
-      (input: Any) => {
-        val data = input.asInstanceOf[java.util.Map[Any, Any]]
-        val keys = new Array[Any](data.size)
-        val values = new Array[Any](data.size)
-        val iter = data.entrySet().iterator()
-        var i = 0
-        while (iter.hasNext) {
-          val entry = iter.next()
-          val (key, value) = (entry.getKey, entry.getValue)
-          keys(i) = if (key != null) {
-            keyConverter.eval(InternalRow.fromSeq(key :: Nil))
-          } else {
-            throw new RuntimeException("Cannot use null as map key!")
-          }
-          values(i) = if (value != null) {
-            valueConverter.eval(InternalRow.fromSeq(value :: Nil))
-          } else {
-            null
+  private lazy val mapCatalystConverter: Any => (Array[Any], Array[Any]) = {
+    val rowBuffer = InternalRow.fromSeq(Array[Any](1))
+    def rowWrapper(data: Any): InternalRow = {
+      rowBuffer.update(0, data)
+      rowBuffer
+    }
+
+    child.dataType match {
+      case ObjectType(cls) if classOf[java.util.Map[_, _]].isAssignableFrom(cls) =>
+        (input: Any) => {
+          val data = input.asInstanceOf[java.util.Map[Any, Any]]
+          val keys = new Array[Any](data.size)
+          val values = new Array[Any](data.size)
+          val iter = data.entrySet().iterator()
+          var i = 0
+          while (iter.hasNext) {
+            val entry = iter.next()
+            val (key, value) = (entry.getKey, entry.getValue)
+            keys(i) = if (key != null) {
+              keyConverter.eval(rowWrapper(key))
+            } else {
+              throw new RuntimeException("Cannot use null as map key!")
+            }
+            values(i) = if (value != null) {
+              valueConverter.eval(rowWrapper(value))
+            } else {
+              null
+            }
+            i += 1
           }
-          i += 1
+          (keys, values)
         }
-        (keys, values)
-      }
 
-    case ObjectType(cls) if classOf[scala.collection.Map[_, _]].isAssignableFrom(cls) =>
-      (input: Any) => {
-        val data = input.asInstanceOf[scala.collection.Map[Any, Any]]
-        val keys = new Array[Any](data.size)
-        val values = new Array[Any](data.size)
-        var i = 0
-        for ((key, value) <- data) {
-          keys(i) = if (key != null) {
-            keyConverter.eval(InternalRow.fromSeq(key :: Nil))
-          } else {
-            throw new RuntimeException("Cannot use null as map key!")
-          }
-          values(i) = if (value != null) {
-            valueConverter.eval(InternalRow.fromSeq(value :: Nil))
-          } else {
-            null
+      case ObjectType(cls) if classOf[scala.collection.Map[_, _]].isAssignableFrom(cls) =>
+        (input: Any) => {
+          val data = input.asInstanceOf[scala.collection.Map[Any, Any]]
+          val keys = new Array[Any](data.size)
+          val values = new Array[Any](data.size)
+          var i = 0
+          for ((key, value) <- data) {
+            keys(i) = if (key != null) {
+              keyConverter.eval(rowWrapper(key))
+            } else {
+              throw new RuntimeException("Cannot use null as map key!")
+            }
+            values(i) = if (value != null) {
+              valueConverter.eval(rowWrapper(value))
+            } else {
+              null
+            }
+            i += 1
           }
-          i += 1
+          (keys, values)
         }
-        (keys, values)
-      }
+    }
   }
 
   override def eval(input: InternalRow): Any = {