[SPARK-49460][SQL] Remove cleanupResource() from EmptyRelationExec

sql/core/src/main/scala/org/apache/spark/sql/execution/EmptyRelationExec.scala

@@ -83,10 +83,13 @@ case class EmptyRelationExec(@transient logical: LogicalPlan) extends LeafExecNo
   }
 
   override protected[sql] def cleanupResources(): Unit = {
-    logical.foreach {
-      case LogicalQueryStage(_, physical) =>
-        physical.cleanupResources()
-      case _ =>
+    // This code path might be executed in executor where `logical` could be null.
+    if (logical != null) {
+      logical.foreach {
+        case LogicalQueryStage(_, physical) =>
+          physical.cleanupResources()
+        case _ =>
+      }
     }
     super.cleanupResources()
   }

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

@@ -1608,6 +1608,47 @@ class AdaptiveQueryExecSuite
     }
   }
 
+  test("SPARK-49460: NPE error in EmptyRelationExec.cleanupResources") {
+    try {
+      spark.sql("create table t1left (a int, b int);")
+      spark.sql("insert into t1left values (1, 1), (2,2), (3,3);")
+      spark.sql("create table t1right (a int, b int);")
+      spark.sql("create table t1empty (a int, b int);")
+      spark.sql("insert into t1right values (2,20), (4, 40);")
+
+      spark.sql("""
+                  |with leftT as (
+                  |  with erp as (
+                  |    select
+                  |      *
+                  |    from
+                  |      t1left
+                  |      join t1empty on t1left.a = t1empty.a
+                  |      join t1right on t1left.a = t1right.a
+                  |  )
+                  |  SELECT
+                  |    CASE
+                  |      WHEN COUNT(*) = 0 THEN 4
+                  |      ELSE NULL
+                  |    END AS a
+                  |  FROM
+                  |    erp
+                  |  HAVING
+                  |    COUNT(*) = 0
+                  |)
+                  |select
+                  |  /*+ MERGEJOIN(t1right) */
+                  |  *
+                  |from
+                  |  leftT
+                  |  join t1right on leftT.a = t1right.a""".stripMargin).collect()
+    } finally {
+      Seq("t1left", "t1right", "t1empty").foreach { table =>
+        spark.sql(s"drop table if exists ${table}")
+      }
+    }
+  }
+
   test("SPARK-35585: Support propagate empty relation through project/filter") {
     withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
       SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {