[BACKPORT-2.1][SPARK-19104][SQL] Lambda variables in ExternalMapToCatalyst should be global

[kiszk]

What changes were proposed in this pull request?

This PR is backport of #18418 to Spark 2.1. This problem is raised by SPARK-21391.

The issue happens in ExternalMapToCatalyst. For example, the following codes create ExternalMapToCatalyst to convert Scala Map to catalyst map format.

val data = Seq.tabulate(10)(i => NestedData(1, Map("key" -> InnerData("name", i + 100))))
val ds = spark.createDataset(data)

The valueConverter in ExternalMapToCatalyst looks like:

if (isnull(lambdavariable(ExternalMapToCatalyst_value52, ExternalMapToCatalyst_value_isNull52, ObjectType(class org.apache.spark.sql.InnerData), true))) null else named_struct(name, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(lambdavariable(ExternalMapToCatalyst_value52, ExternalMapToCatalyst_value_isNull52, ObjectType(class org.apache.spark.sql.InnerData), true)).name, true), value, assertnotnull(lambdavariable(ExternalMapToCatalyst_value52, ExternalMapToCatalyst_value_isNull52, ObjectType(class org.apache.spark.sql.InnerData), true)).value)

There is a CreateNamedStruct expression (named_struct) to create a row of InnerData.name and InnerData.value that are referred by ExternalMapToCatalyst_value52.

Because ExternalMapToCatalyst_value52 are local variable, when CreateNamedStruct splits expressions to individual functions, the local variable can't be accessed anymore.

How was this patch tested?

Added a new test suite into DatasetPrimitiveSuite

[kiszk]

Made a mistake

[SparkQA]

Test build #79589 has finished for PR 18626 at commit a6985cf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.