Fix test error

Author: wangyum

sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala

@@ -280,9 +280,9 @@ class DataFrameJoinSuite extends QueryTest
       plan match {
         // SPARK-34178: we can't match the plan before the fix due to
         // the right side plan doesn't contains dataset id.
-        case Join(
-          LogicalPlanWithDatasetId(_, leftId),
-          LogicalPlanWithDatasetId(_, rightId), _, _, _) =>
+        case Project(_, Join(
+          Project(_, LogicalPlanWithDatasetId(_, leftId)),
+          Project(_, LogicalPlanWithDatasetId(_, rightId)), _, _, _)) =>
           assert(leftId === rightId)
       }
     }

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

@@ -1057,7 +1057,7 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
     val pythonEvals = collect(joinNode.get) {
       case p: BatchEvalPythonExec => p
     }
-    assert(pythonEvals.size == 2)
+    assert(pythonEvals.size == 4)
 
     checkAnswer(df, Row(1, 2, 1, 2) :: Nil)
   }

sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala

@@ -143,9 +143,10 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
     val schema = new StructType().add("k", IntegerType).add("v", StringType)
     val smallDF = spark.createDataFrame(rdd, schema)
     val df = spark.range(10).join(broadcast(smallDF), col("k") === col("id"))
-    assert(df.queryExecution.executedPlan.find(p =>
-      p.isInstanceOf[WholeStageCodegenExec] &&
-        p.asInstanceOf[WholeStageCodegenExec].child.isInstanceOf[BroadcastHashJoinExec]).isDefined)
+    val broadcastHashJoin = df.queryExecution.executedPlan.find {
+      case WholeStageCodegenExec(ProjectExec(_, _: BroadcastHashJoinExec)) => true
+    }
+    assert(broadcastHashJoin.isDefined)
     assert(df.collect() === Array(Row(1, 1, "1"), Row(1, 1, "1"), Row(2, 2, "2")))
   }
 
@@ -187,7 +188,7 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
     // test one join with non-unique key from build side
     val joinNonUniqueDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2" % 3, "full_outer")
     assert(joinNonUniqueDF.queryExecution.executedPlan.collect {
-      case WholeStageCodegenExec(_ : ShuffledHashJoinExec) => true
+      case WholeStageCodegenExec(ProjectExec(_, _: ShuffledHashJoinExec)) => true
     }.size === 1)
     checkAnswer(joinNonUniqueDF, Seq(Row(0, 0), Row(0, 3), Row(0, 6), Row(0, 9), Row(1, 1),
       Row(1, 4), Row(1, 7), Row(2, 2), Row(2, 5), Row(2, 8), Row(3, null), Row(4, null)))
@@ -196,7 +197,7 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
     val joinWithNonEquiDF = df1.join(df2.hint("SHUFFLE_HASH"),
       $"k1" === $"k2" % 3 && $"k1" + 3 =!= $"k2", "full_outer")
     assert(joinWithNonEquiDF.queryExecution.executedPlan.collect {
-      case WholeStageCodegenExec(_ : ShuffledHashJoinExec) => true
+      case WholeStageCodegenExec(ProjectExec(_, _: ShuffledHashJoinExec)) => true
     }.size === 1)
     checkAnswer(joinWithNonEquiDF, Seq(Row(0, 0), Row(0, 6), Row(0, 9), Row(1, 1),
       Row(1, 7), Row(2, 2), Row(2, 8), Row(3, null), Row(4, null), Row(null, 3), Row(null, 4),