[SPARK-37829][SQL] Dataframe.joinWith outer-join should return a null value for unmatched row

[kings129]

What changes were proposed in this pull request?

When doing an outer join with joinWith on DataFrames, unmatched rows return Row objects with null fields instead of a single null value. This is not a expected behavior, and it's a regression introduced in this commit.
This pull request aims to fix the regression, note this is not a full rollback of the commit, do not add back "schema" variable.

case class ClassData(a: String, b: Int)
val left = Seq(ClassData("a", 1), ClassData("b", 2)).toDF
val right = Seq(ClassData("x", 2), ClassData("y", 3)).toDF

left.joinWith(right, left("b") === right("b"), "left_outer").collect

Wrong results (current behavior):    Array(([a,1],[null,null]), ([b,2],[x,2]))
Correct results:                     Array(([a,1],null), ([b,2],[x,2]))

Why are the changes needed?

We need to address the regression mentioned above. It results in unexpected behavior changes in the Dataframe joinWith API between versions 2.4.8 and 3.0.0+. This could potentially cause data correctness issues for users who expect the old behavior when using Spark 3.0.0+.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added unit test (use the same test in previous closed pull request, credit to Clément de Groc)
Run sql-core and sql-catalyst submodules locally with ./build/mvn clean package -pl sql/core,sql/catalyst

[hvanhovell]

cc @zhenlineo since you are working on this on the connect side.

[kings129]

cc @cloud-fan @viirya for review, thanks!

[cloud-fan]

it's kind of we push down the null check to the children deserializers. Why is the serializer fine?

[kings129]

This change is intended to create a deserializer type newinstance(class scala.Tuple*) that can convert to a single null value. This behavior is the same as before the commit introduced the regression.
Regarding the serializer, in the new unit test added in this pull request, when the tuple is not null, named_struct is created for each element, and null is handled there.

if (isnull(input[0, scala.Tuple2, true])) null else named_struct(_1, if (isnull(input[0, scala.Tuple2, true]._1)) null else named_struct(a, if (input[0, scala.Tuple2, true]._1.isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(input[0, scala.Tuple2, true]._1, 0, a), StringType, ObjectType(class java.lang.String)), true, false, true), b, assertnotnull(validateexternaltype(getexternalrowfield(input[0, scala.Tuple2, true]._1, 1, b), IntegerType, ObjectType(class java.lang.Integer)).intValue)) AS _1#18, _2, if (isnull(input[0, scala.Tuple2, true]._2)) null else named_struct(a, if (input[0, scala.Tuple2, true]._2.isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(input[0, scala.Tuple2, true]._2, 0, a), StringType, ObjectType(class java.lang.String)), true, false, true), b, assertnotnull(validateexternaltype(getexternalrowfield(input[0, scala.Tuple2, true]._2, 1, b), IntegerType, ObjectType(class java.lang.Integer)).intValue)) AS _2#19)

[kings129]

@cloud-fan does my comment answer your question? PTAL, thanks!

[cloud-fan]

It looks correct to me to add the null check for the children deserializers. But I don't quite understand why this PR removes the outermost null check. After looking at the code, I think it doesn't matter, as the outermost null check will be removed anyway: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala#L274

Since this is unrelated to this PR, let's not touch it. If you do want to fix it (adding null check and removing it later is useless), let's fix the serializer as well.

[kings129]

@cloud-fan, thanks for the explanation! You're right; it doesn't matter whether to keep the outermost null check. (null check for deserializer was also added in refactor commit)

I also prefer making minimal changes to fix the target issue. I added back the outermost null check for the deserializer.

[cloud-fan]

thanks, merging to master/3.4!

[cloud-fan]

@kings129 can you open a new PR for branch 3.3? Thanks!

[kings129]

@kings129 can you open a new PR for branch 3.3? Thanks!

Thanks for the quick review, @cloud-fan!
Yes, here is the pull request for branch 3.3: #40858