[SPARK-56395][SQL][FOLLOWUP] Generalize nullability test to cover the nondeterministic ranking path

Generalize the LEFT OUTER nullability regression test in RewriteNearestByJoinSuite
from an Aggregate-only assertion to a whole-plan integrity walk, and add a
nondeterministic-ranking case alongside the deterministic one across both Inner
and LeftOuter.

The widened ranking reference lands in the Aggregate on the deterministic path but
in a `__ranking__` Project (above the Join) on the nondeterministic path. The old
assertion only inspected Aggregate expressions, so a regression isolated to the
nondeterministic branch would not have been caught. No framework check guards this
class of bug either -- LogicalPlanIntegrity compares types `asNullable` and schemas
via `equalsIgnoreNullability` -- so this test is the only guard.

Verified the new coverage has teeth: reverting just the nondeterministic alias to
the un-widened expression fails the test specifically on LeftOuter/nondeterministic
while the deterministic case still passes.

Co-authored-by: Isaac

Author: cloud-fan

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteNearestByJoinSuite.scala

@@ -156,27 +156,35 @@ class RewriteNearestByJoinSuite extends PlanTest {
   }
 
   test("SPARK-56395: LEFT OUTER rewrite keeps right-side nullability consistent with its child") {
-    // A LEFT OUTER NEAREST BY widens the synthetic join's right-side columns to nullable. The
-    // Aggregate built on top of that join references those columns (in the `_matches` struct and
-    // the ranking), so each reference must carry the widened nullability -- otherwise the
-    // rewritten plan declares a column non-nullable while its child produces it as nullable, an
-    // internal inconsistency. INNER does not widen the right side, so it stays a no-op.
+    // A LEFT OUTER NEAREST BY widens the synthetic join's right-side columns to nullable. Every
+    // operator built on top of that join that references those columns (the `_matches` struct,
+    // the ranking) must carry the widened nullability -- otherwise the rewritten plan declares a
+    // column non-nullable while its child produces it as nullable, an internal inconsistency that
+    // no framework check catches (`LogicalPlanIntegrity` compares types `asNullable` and schemas
+    // `equalsIgnoreNullability`), so this assertion is the only guard. INNER does not widen the
+    // right side, so it stays a no-op.
     //
     // The right-side columns are declared non-nullable here: that is what makes LEFT OUTER's
     // widening observable (with nullable columns the widening is a no-op and the bug is hidden).
+    // The ranking is exercised both deterministic (the reference lands directly in the Aggregate)
+    // and nondeterministic (the rule pre-materializes it into a `__ranking__` Project above the
+    // Join), so the widening is checked wherever the reference ends up.
     val left = LocalRelation($"a".int, $"b".int)
     val right = LocalRelation(
       AttributeReference("x", IntegerType, nullable = false)(),
       AttributeReference("y", IntegerType, nullable = false)())
-    Seq(Inner -> false, LeftOuter -> true).foreach { case (joinType, rightNullable) =>
+    val rankings = Seq(
+      "deterministic" -> (left.output(0) + right.output(0)),
+      "nondeterministic" -> (Rand(Literal(0L)) + right.output(0)))
+    for ((joinType, rightNullable) <- Seq(Inner -> false, LeftOuter -> true);
+         (label, ranking) <- rankings) {
       val query = NearestByJoin(
         left, right, joinType, approx = true, numResults = 1,
-        rankingExpression = left.output(0) + right.output(0),
+        rankingExpression = ranking,
         direction = NearestBySimilarity)
 
       val rewritten = RewriteNearestByJoin(query.analyze)
       val join = rewritten.collect { case j: Join => j }.head
-      val aggregate = rewritten.collect { case a: Aggregate => a }.head
 
       // Sanity-check the fixture: the synthetic join widens its right-side output to nullable
       // iff it is LEFT OUTER. (`join.right` is the right relation as it appears in the rewritten
@@ -186,16 +194,23 @@ class RewriteNearestByJoinSuite extends PlanTest {
       assert(joinRightOutput.nonEmpty)
       assert(joinRightOutput.forall(_.nullable == rightNullable))
 
-      // Every attribute the Aggregate references that its child (the join) produces must agree
-      // on nullability with the child -- this is exactly what the fix corrects for LEFT OUTER.
-      val childNullability = aggregate.child.output.map(a => a.exprId -> a.nullable).toMap
-      aggregate.aggregateExpressions.foreach(_.foreach {
-        case ref: AttributeReference if childNullability.contains(ref.exprId) =>
-          assert(ref.nullable == childNullability(ref.exprId),
-            s"$joinType: ${ref.name}#${ref.exprId.id} declared nullable=${ref.nullable} " +
-              s"but its child produces nullable=${childNullability(ref.exprId)}")
-        case _ =>
-      })
+      // Whole-plan integrity: at every operator, an attribute reference whose ExprId is produced
+      // by one of that operator's children must agree with the child on nullability -- this is
+      // exactly what the fix corrects for LEFT OUTER. Walking the whole plan (rather than just
+      // the Aggregate) also covers the `__ranking__` Project that the nondeterministic path
+      // inserts above the Join, where the widened ranking reference lands.
+      rewritten.foreach { node =>
+        val childNullability =
+          node.children.flatMap(_.output).map(a => a.exprId -> a.nullable).toMap
+        node.expressions.foreach(_.foreach {
+          case ref: AttributeReference if childNullability.contains(ref.exprId) =>
+            assert(ref.nullable == childNullability(ref.exprId),
+              s"$joinType/$label: ${ref.name}#${ref.exprId.id} declared " +
+                s"nullable=${ref.nullable} but its child produces " +
+                s"nullable=${childNullability(ref.exprId)}")
+          case _ =>
+        })
+      }
     }
   }