[SPARK-47927][SQL][FOLLOWUP] fix ScalaUDF output nullability

[cloud-fan]

What changes were proposed in this pull request?

This is a followup of #46156 , to fix the wrong nullability of ScalaUDF output.

Why are the changes needed?

fix nullability

Does this PR introduce any user-facing change?

no

How was this patch tested?

new test

Was this patch authored or co-authored using generative AI tooling?

no

[cloud-fan]

cc @eejbyfeldt @HyukjinKwon

[cloud-fan]

This rule adds null handling for ScalaUDF, so even if the ScalaUDF's nullable is false, the final expression (if-else) can also return null, so we need to run UpdateAttributeNullability after it.

[eejbyfeldt]

But we also need to run it before? Or how could we correctly handle null inputs to a udf if we do not have the correct nullability for the inputs?

Seems to me like one should be able to write a test case involving an outer join that shows that the proposed code will not handle such nullable input correctly.

[eejbyfeldt]

Feels like this does not address the root issue. To me it seems problematic that nullable for ScalaUDF does not correctly reflect if the expression is nullable or not. Could that not cause bugs in other parts of the code?

Would it be better to address this issue by changing the nullable implementation? Something like #47085

[cloud-fan]

@eejbyfeldt I did consider this approach before. However, the null check happens outside of ScalaUDF. If we do invoke the UDF, ScalaUDF.nullable = false means the UDF result won't be null.

[eejbyfeldt]

@eejbyfeldt I did consider this approach before. However, the null check happens outside of ScalaUDF. If we do invoke the UDF, ScalaUDF.nullable = false means the UDF result won't be null.

Could you expand on how to trigger this error? I am not sure I am following. (But I do agree that it feels a bit off the the implementation of nullable does not match implementation of the execution)

Here is a code that still produces incorrect result with this branch (and master and prior to #46156 ):

scala> val f = udf[Int, Int](_ + 1)
scala> sc.parallelize(Seq(Some(1), None)).toDS.select(f($"value").as("t"), f($"t")).show()
+----+-----------------------------+
|   t|UDF(lateralAliasReference(t))|
+----+-----------------------------+
|   2|                            3|
|NULL|                            1|
+----+-----------------------------+

To solve (while keeping the ScalaUDF.nullable implementation) this we would probably need to make the update nullability batch be something like

    Batch("UpdateNullability", fixedPoint,
      HandleNullInputsForUDF,
      UpdateAttributeNullability),

[cloud-fan]

@eejbyfeldt good catch! I've fixed it and added a test

[cloud-fan]

thanks for the reviews, merging to master/3.5/3.4!