[SPARK-50798][SQL][FOLLOWUP] Normalize expressions with random seed

### What changes were proposed in this pull request?
This PR introduces normalization of expressions with random seed.

### Why are the changes needed?
Single-pass and fixed-point need to produce same plans after normalization, so we need to normalize non-deterministic expressions.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Added a test case to `NormalizePlanSuite`.

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

Closes #49623 from mihailotim-db/mihailotim-db/normalization_followup_nondeterministic.

Authored-by: Mihailo Timotic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: mihailotim-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/NormalizePlan.scala

@@ -41,10 +41,12 @@ object NormalizePlan extends PredicateHelper {
   def normalizeExpressions(plan: LogicalPlan): LogicalPlan = {
     val withNormalizedRuntimeReplaceable = normalizeRuntimeReplaceable(plan)
     withNormalizedRuntimeReplaceable transformAllExpressions {
-      case c: CommonExpressionDef =>
-        c.copy(id = new CommonExpressionId(id = 0))
-      case c: CommonExpressionRef =>
-        c.copy(id = new CommonExpressionId(id = 0))
+      case commonExpressionDef: CommonExpressionDef =>
+        commonExpressionDef.copy(id = new CommonExpressionId(id = 0))
+      case commonExpressionRef: CommonExpressionRef =>
+        commonExpressionRef.copy(id = new CommonExpressionId(id = 0))
+      case expressionWithRandomSeed: ExpressionWithRandomSeed =>
+        expressionWithRandomSeed.withNewSeed(0)
     }
   }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/NormalizePlanSuite.scala

@@ -17,8 +17,11 @@
 
 package org.apache.spark.sql.catalyst.plans
 
+import scala.util.Random
+
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.expressions.{
   AssertTrue,
@@ -100,6 +103,18 @@ class NormalizePlanSuite extends SparkFunSuite with SQLConfHelper {
     assert(NormalizePlan(baselinePlanDef) == NormalizePlan(testPlanDef))
   }
 
+  test("Normalize non-deterministic expressions") {
+    val random = new Random()
+    val baselineExpression = rand(random.nextLong())
+    val testExpression = rand(random.nextLong())
+
+    val baselinePlan = LocalRelation().select(baselineExpression)
+    val testPlan = LocalRelation().select(testExpression)
+
+    assert(baselinePlan != testPlan)
+    assert(NormalizePlan(baselinePlan) == NormalizePlan(testPlan))
+  }
+
   private def setTimezoneForAllExpression(plan: LogicalPlan): LogicalPlan = {
     plan.transformAllExpressions {
       case e: TimeZoneAwareExpression if e.timeZoneId.isEmpty =>