[SPARK-37829][SQL] Add if(isnull ...) check for DataFrame.joinWith

Wrap tuple fields deserializers in null checks when calling on
DataFrames as top-level rows are not nullable and won't propagate
null values.

Author: cdegroc

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala

@@ -77,6 +77,28 @@ object RowEncoder {
       ClassTag(cls))
   }
 
+  /**
+   * Returns an ExpressionEncoder allowing null top-level rows.
+   * @param exprEnc an ExpressionEncoder[Row].
+   * @return an ExpressionEncoder[Row] whom deserializer supports null values.
+   *
+   * @see SPARK-37829
+   */
+  private[sql] def nullSafe(exprEnc: ExpressionEncoder[Row]): ExpressionEncoder[Row] = {
+    val newDeserializerInput = GetColumnByOrdinal(0, exprEnc.objSerializer.dataType)
+    val newDeserializer: Expression = if (exprEnc.objSerializer.nullable) {
+      If(
+        IsNull(newDeserializerInput),
+        Literal.create(null, exprEnc.objDeserializer.dataType),
+        exprEnc.objDeserializer)
+    } else {
+      exprEnc.objDeserializer
+    }
+    exprEnc.copy(
+      objDeserializer = newDeserializer
+    )
+  }
+
   private def serializerFor(
       inputObject: Expression,
       inputType: DataType): Expression = inputType match {

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -1173,8 +1173,20 @@ class Dataset[T] private[sql](
       joined = resolveSelfJoinCondition(joined)
     }
 
-    implicit val tuple2Encoder: Encoder[(T, U)] =
-      ExpressionEncoder.tuple(this.exprEnc, other.exprEnc)
+    // SPARK-37829: an outer-join requires the null semantics to represent missing keys.
+    // As we might be running on DataFrames, we need a custom encoder that will properly
+    // handle null top-level Rows.
+    def nullSafe[V](exprEnc: ExpressionEncoder[V]): ExpressionEncoder[V] = {
+      if (exprEnc.clsTag.runtimeClass != classOf[Row]) {
+        exprEnc
+      } else {
+        RowEncoder.nullSafe(exprEnc.asInstanceOf[ExpressionEncoder[Row]])
+          .asInstanceOf[ExpressionEncoder[V]]
+      }
+    }
+    implicit val tuple2Encoder: Encoder[(T, U)] = ExpressionEncoder.tuple(
+      nullSafe(this.exprEnc), nullSafe(other.exprEnc)
+    )
 
     val leftResultExpr = {
       if (!this.exprEnc.isSerializedAsStructForTopLevel) {