Revert "[SPARK-44801][SQL][UI] Capture analyzing failed queries in Listener and UI"

### What changes were proposed in this pull request?
In the PR, I propose to revert the commit: apache@1f92995

### Why are the changes needed?
To fix the test suite `StringFunctionsSuite`.

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

### How was this patch tested?
By running the affected test suite:
```
$ build/sbt "test:testOnly *.StringFunctionsSuite"
```

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

Closes apache#42589 from MaxGekk/revert-yaooqinn-SPARK-44801.

Authored-by: Max Gekk <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: MaxGekk

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

@@ -63,17 +63,7 @@ class QueryExecution(
   // TODO: Move the planner an optimizer into here from SessionState.
   protected def planner = sparkSession.sessionState.planner
 
-  def assertAnalyzed(): Unit = {
-    try {
-      analyzed
-    } catch {
-      case e: AnalysisException =>
-        // Because we do eager analysis for Dataframe, there will be no execution created after
-        // AnalysisException occurs. So we need to explicitly create a new execution to post
-        // start/end events to notify the listener and UI components.
-        SQLExecution.withNewExecutionId(this, Some("analyze"))(throw e)
-    }
-  }
+  def assertAnalyzed(): Unit = analyzed
 
   def assertSupported(): Unit = {
     if (sparkSession.sessionState.conf.isUnsupportedOperationCheckEnabled) {

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

@@ -20,11 +20,8 @@ package org.apache.spark.sql.execution
 import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Future => JFuture}
 import java.util.concurrent.atomic.AtomicLong
 
-import scala.util.control.NonFatal
-
 import org.apache.spark.{ErrorMessageFormat, SparkThrowable, SparkThrowableHelper}
 import org.apache.spark.SparkContext.{SPARK_JOB_DESCRIPTION, SPARK_JOB_INTERRUPT_ON_CANCEL}
-import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.{SPARK_DRIVER_PREFIX, SPARK_EXECUTOR_PREFIX}
 import org.apache.spark.internal.config.Tests.IS_TESTING
 import org.apache.spark.sql.SparkSession
@@ -33,7 +30,7 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.StaticSQLConf.SQL_EVENT_TRUNCATE_LENGTH
 import org.apache.spark.util.Utils
 
-object SQLExecution extends Logging {
+object SQLExecution {
 
   val EXECUTION_ID_KEY = "spark.sql.execution.id"
   val EXECUTION_ROOT_ID_KEY = "spark.sql.execution.root.id"
@@ -119,15 +116,6 @@ object SQLExecution extends Logging {
         var ex: Option[Throwable] = None
         val startTime = System.nanoTime()
         try {
-          val planInfo = try {
-            SparkPlanInfo.fromSparkPlan(queryExecution.executedPlan)
-          } catch {
-            case NonFatal(e) =>
-              logDebug("Failed to generate SparkPlanInfo", e)
-              // If the queryExecution already failed before this, we are not able to generate the
-              // the plan info, so we use and empty graphviz node to make the UI happy
-              SparkPlanInfo.EMPTY
-          }
           sc.listenerBus.post(SparkListenerSQLExecutionStart(
             executionId = executionId,
             rootExecutionId = Some(rootExecutionId),
@@ -136,7 +124,7 @@ object SQLExecution extends Logging {
             physicalPlanDescription = queryExecution.explainString(planDescriptionMode),
             // `queryExecution.executedPlan` triggers query planning. If it fails, the exception
             // will be caught and reported in the `SparkListenerSQLExecutionEnd`
-            sparkPlanInfo = planInfo,
+            sparkPlanInfo = SparkPlanInfo.fromSparkPlan(queryExecution.executedPlan),
             time = System.currentTimeMillis(),
             modifiedConfigs = redactedConfigs,
             jobTags = sc.getJobTags()

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

@@ -76,6 +76,4 @@ private[execution] object SparkPlanInfo {
       metadata,
       metrics)
   }
-
-  final lazy val EMPTY: SparkPlanInfo = new SparkPlanInfo("", "", Nil, Map.empty, Nil)
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/ui/UISeleniumSuite.scala

@@ -32,31 +32,10 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser {
 
   private var spark: SparkSession = _
 
-  private def creatSparkSessionWithUI: SparkSession = SparkSession.builder()
-    .master("local[1,1]")
-    .appName("sql ui test")
-    .config("spark.ui.enabled", "true")
-    .config("spark.ui.port", "0")
-    .getOrCreate()
-
   implicit val webDriver: HtmlUnitDriver = new HtmlUnitDriver {
     getWebClient.setCssErrorHandler(new SparkUICssErrorHandler)
   }
 
-  private def findErrorMessageOnSQLUI(): List[String] = {
-    val webUrl = spark.sparkContext.uiWebUrl
-    assert(webUrl.isDefined, "please turn on spark.ui.enabled")
-    go to s"${webUrl.get}/SQL"
-    findAll(cssSelector("""#failed-table td .stacktrace-details""")).map(_.text).toList
-  }
-
-  private def findExecutionIDOnSQLUI(): Int = {
-    val webUrl = spark.sparkContext.uiWebUrl
-    assert(webUrl.isDefined, "please turn on spark.ui.enabled")
-    go to s"${webUrl.get}/SQL"
-    findAll(cssSelector("""#failed-table td""")).map(_.text).toList.head.toInt
-  }
-
   override def afterAll(): Unit = {
     try {
       webDriver.quit()
@@ -75,32 +54,21 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser {
   }
 
   test("SPARK-44737: Should not display json format errors on SQL page for non-SparkThrowables") {
-    spark = creatSparkSessionWithUI
+    spark = SparkSession.builder()
+      .master("local[1,1]")
+      .appName("sql ui test")
+      .config("spark.ui.enabled", "true")
+      .config("spark.ui.port", "0")
+      .getOrCreate()
 
     intercept[Exception](spark.sql("SET mapreduce.job.reduces = 0").isEmpty)
     eventually(timeout(10.seconds), interval(100.milliseconds)) {
-      val sd = findErrorMessageOnSQLUI()
+      val webUrl = spark.sparkContext.uiWebUrl
+      assert(webUrl.isDefined, "please turn on spark.ui.enabled")
+      go to s"${webUrl.get}/SQL"
+      val sd = findAll(cssSelector("""#failed-table td .stacktrace-details""")).map(_.text).toList
       assert(sd.size === 1, "SET mapreduce.job.reduces = 0 shall fail")
       assert(sd.head.startsWith("java.lang.IllegalArgumentException:"))
     }
   }
-
-  test("SPARK-44801: Analyzer failure shall show the query in failed table") {
-    spark = creatSparkSessionWithUI
-
-    intercept[Exception](spark.sql("SELECT * FROM I_AM_A_INVISIBLE_TABLE").isEmpty)
-    eventually(timeout(10.seconds), interval(100.milliseconds)) {
-      val sd = findErrorMessageOnSQLUI()
-      assert(sd.size === 1, "Analyze fail shall show the query in failed table")
-      assert(sd.head.startsWith("[TABLE_OR_VIEW_NOT_FOUND]"))
-
-      val id = findExecutionIDOnSQLUI()
-      // check query detail page
-      go to s"${spark.sparkContext.uiWebUrl.get}/SQL/execution/?id=$id"
-      val planDot = findAll(cssSelector(""".dot-file""")).map(_.text).toList
-      assert(planDot.head.startsWith("digraph G {"))
-      val planDetails = findAll(cssSelector("""#physical-plan-details""")).map(_.text).toList
-      assert(planDetails.head.contains("TABLE_OR_VIEW_NOT_FOUND"))
-    }
-  }
 }