[SPARK-35653][SQL] Fix CatalystToExternalMap interpreted path fails for Map with case classes as keys or values

[eejbyfeldt]

What changes were proposed in this pull request?

Use the key/value LambdaFunction to convert the elements instead of
using CatalystTypeConverters.createToScalaConverter. This is how it is
done in MapObjects and that correctly handles Arrays with case classes.

Why are the changes needed?

Before these changes the added test cases would fail with the following:

[info] - encode/decode for map with case class as value: Map(1 -> IntAndString(1,a)) (interpreted path) *** FAILED *** (64 milliseconds)
[info]   Encoded/Decoded data does not match input data
[info]   
[info]   in:  Map(1 -> IntAndString(1,a))
[info]   out: Map(1 -> [1,a])
[info]   types: scala.collection.immutable.Map$Map1 [info]   
[info]   Encoded Data: [org.apache.spark.sql.catalyst.expressions.UnsafeMapData@5ecf5d9e]
[info]   Schema: value#823
[info]   root
[info]   -- value: map (nullable = true)
[info]       |-- key: integer
[info]       |-- value: struct (valueContainsNull = true)
[info]       |    |-- i: integer (nullable = false)
[info]       |    |-- s: string (nullable = true)
[info]   
[info]   
[info]   fromRow Expressions:
[info]   catalysttoexternalmap(lambdavariable(CatalystToExternalMap_key, IntegerType, false, 178), lambdavariable(CatalystToExternalMap_key, IntegerType, false, 178), lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179), if (isnull(lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179))) null else newInstance(class org.apache.spark.sql.catalyst.encoders.IntAndString), input[0, map<int,struct<i:int,s:string>>, true], interface scala.collection.immutable.Map
[info]   :- lambdavariable(CatalystToExternalMap_key, IntegerType, false, 178)
[info]   :- lambdavariable(CatalystToExternalMap_key, IntegerType, false, 178)
[info]   :- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179)
[info]   :- if (isnull(lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179))) null else newInstance(class org.apache.spark.sql.catalyst.encoders.IntAndString)
[info]   :  :- isnull(lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179))
[info]   :  :  +- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179)
[info]   :  :- null
[info]   :  +- newInstance(class org.apache.spark.sql.catalyst.encoders.IntAndString)
[info]   :     :- assertnotnull(lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179).i)
[info]   :     :  +- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179).i
[info]   :     :     +- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179)
[info]   :     +- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179).s.toString
[info]   :        +- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179).s
[info]   :           +- lambdavariable(CatalystToExternalMap_value, StructField(i,IntegerType,false), StructField(s,StringType,true), true, 179)
[info]   +- input[0, map<int,struct<i:int,s:string>>, true] (ExpressionEncoderSuite.scala:627)

So using a map with cases classes for keys or values and using the interpreted path would incorrect deserialize data from the catalyst representation.

Does this PR introduce any user-facing change?

Yes, it fixes the bug.

How was this patch tested?

Existing and new unit tests in the ExpressionEncoderSuite

[dongjoon-hyun]

ok to test

[dongjoon-hyun]

Thank you for making a PR, @eejbyfeldt .

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43874/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43874/

[SparkQA]

Test build #139352 has finished for PR 32783 at commit e95d7f4.

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

[maropu]

We tend to use while for perf-intensive code. The proposed code does not cause any perf overhead?

[eejbyfeldt]

I guess I am not sure whether it does or not. But I updated the PR to use a while instead, to be sure. Now the style is also more similar to what was there before.

[maropu]

The latest one looks fine. Thanks ;)

[maropu]

Nice catch, @eejbyfeldt and thank you for your contribution.

[maropu]

cc: @viirya

[maropu]

Looks fine if the tests pass.

[SparkQA]

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43979/

[SparkQA]

Test build #139456 has finished for PR 32783 at commit 9cc2484.

• This patch fails PySpark pip packaging tests.
• This patch merges cleanly.
• This patch adds no public classes.

[eejbyfeldt]

I would think the test failure was caused by the outage of python infrastructure (https://status.python.org/) and not my code. How can I retest this branch?

[maropu]

retest this please

[maropu]

I would think the test failure was caused by the outage of python infrastructure (https://status.python.org/) and not my code. How can I retest this branch?

Yea, the PR looks fine cuz the GA tests passed.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44038/

[SparkQA]

Test build #139513 has finished for PR 32783 at commit 9cc2484.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44038/

[viirya]

StructConverter converts from Catalyst struct to a Row. Will it be a behavior change? Although it is Catalyst expression.

[eejbyfeldt]

StructConverter converts from Catalyst struct to a Row. Will it be a behavior change? Although it is Catalyst expression.

From my understanding this is exactly the bug being fixed. So there will be behavior change. But the behavior is changed such that the interpreted path and the code gen path has the same behavior. The thinking being that the old behavior was undesired and incorrect.

My understanding of the failure the test cases (if added without the patch):

[info] - encode/decode for map with case class as value: Map(1 -> IntAndString(1,a)) (interpreted path) *** FAILED *** (64 milliseconds)
[info]   Encoded/Decoded data does not match input data
[info]   
[info]   in:  Map(1 -> IntAndString(1,a))
[info]   out: Map(1 -> [1,a])

Is that the value of the Map was converted to an [1,a] which I believe is an InternalRow instead of the expected IntAndString. This sounds like the change you are mentioning?

The reason I believe my change is the correct is that using the key/valueLambdaFunction is since this is how it is done inside MapObjects:
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala#L820-L826

Please correct me if I am wrong about anything or misunderstood the question as I am new to the code base.

[viirya]

I think the new behavior looks more correct. Just wondering how the change affects developers or users. Usually for user-facing change, we can document in migration guide. This is per Catalyst expression change, not sure if migration guide is suitable too.

cc @cloud-fan

[cloud-fan]

LGTM. I believe this a bug fix instead of a breaking change.

[viirya]

Thanks @eejbyfeldt for your first contribution! Added you into contributor list on JIRA and assigned this ticket to you. Welcome to Spark community.

Thanks @maropu @cloud-fan for review. Merging to master, 3.1, 3.0.

[maropu]

Thank you, all~