[SPARK-48929] Fix view internal error and clean up parser exception c…

…ontext

### What changes were proposed in this pull request?

Replace INVALID_VIEW_TEXT (SQLSTATE: XX000) error-condition which is forced when view expansion fails to parse with the underlying error.
The reason being that the underlying error has important information AND it is much more likely that view expansion fails because of a behavior change in the parser than some nefarious action or simple corruption.
For example the recent removal of `!` and `NOT` equivalence fails old view on on Spark 4.0 if they exploit the unsupported syntax.

As part of this change we also include the Query context summary in the ParserException processing.
This will expose the view name along side the view query error message.

### Why are the changes needed?

Better error handling improving ability of customers to do root cause analysis.

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

We are now giving more error context in case of parer errors and specifically in case of parser errors in view expansion

### How was this patch tested?

Existing QA

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

No

Closes #47405 from srielau/SPARK-48929-Fix-view-intrenal-error.

Authored-by: Serge Rielau <serge@rielau.com>
Signed-off-by: Gengliang Wang <gengliang@apache.org>

common/utils/src/main/resources/error/error-conditions.json

@@ -2983,12 +2983,6 @@
     ],
     "sqlState" : "22023"
   },
-  "INVALID_VIEW_TEXT" : {
-    "message" : [
-      "The view <viewName> cannot be displayed due to invalid view text: <viewText>. This may be caused by an unauthorized modification of the view or an incorrect query syntax. Please check your query syntax and verify that the view has not been tampered with."
-    ],
-    "sqlState" : "XX000"
-  },
   "INVALID_WHERE_CONDITION" : {
     "message" : [
       "The WHERE condition <condition> contains invalid expressions: <expressionList>.",

sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/parsers.scala

@@ -249,20 +249,27 @@ class ParseException private(
   override def getMessage: String = {
     val builder = new StringBuilder
     builder ++= "\n" ++= message
-    start match {
-      case Origin(Some(l), Some(p), _, _, _, _, _, _, _) =>
-        builder ++= s" (line $l, pos $p)\n"
-        command.foreach { cmd =>
-          val (above, below) = cmd.split("\n").splitAt(l)
-          builder ++= "\n== SQL ==\n"
-          above.foreach(builder ++= _ += '\n')
-          builder ++= (0 until p).map(_ => "-").mkString("") ++= "^^^\n"
-          below.foreach(builder ++= _ += '\n')
-        }
-      case _ =>
-        command.foreach { cmd =>
-          builder ++= "\n== SQL ==\n" ++= cmd
-        }
+    if (queryContext.nonEmpty) {
+      builder ++= "\n"
+      queryContext.foreach { ctx =>
+        builder ++= ctx.summary()
+      }
+    } else {
+      start match {
+        case Origin(Some(l), Some(p), _, _, _, _, _, _, _) =>
+          builder ++= s" (line $l, pos $p)\n"
+          command.foreach { cmd =>
+            val (above, below) = cmd.split("\n").splitAt(l)
+            builder ++= "\n== SQL ==\n"
+            above.foreach(builder ++= _ += '\n')
+            builder ++= (0 until p).map(_ => "-").mkString("") ++= "^^^\n"
+            below.foreach(builder ++= _ += '\n')
+          }
+        case _ =>
+          command.foreach { cmd =>
+            builder ++= "\n== SQL ==\n" ++= cmd
+          }
+      }
     }
     builder.toString
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -36,9 +36,9 @@ import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder
 import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, Expression, ExpressionInfo, NamedExpression, UpCast}
-import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException, ParserInterface}
+import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParserInterface}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, SubqueryAlias, View}
-import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
+import org.apache.spark.sql.catalyst.trees.CurrentOrigin
 import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, StringUtils}
 import org.apache.spark.sql.connector.catalog.CatalogManager
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
@@ -962,19 +962,14 @@ class SessionCatalog(
       throw SparkException.internalError("Invalid view without text.")
     }
     val viewConfigs = metadata.viewSQLConfigs
-    val origin = Origin(
+    val origin = CurrentOrigin.get.copy(
       objectType = Some("VIEW"),
       objectName = Some(metadata.qualifiedName)
     )
     val parsedPlan = SQLConf.withExistingConf(View.effectiveSQLConf(viewConfigs, isTempView)) {
-      try {
         CurrentOrigin.withOrigin(origin) {
           parser.parseQuery(viewText)
         }
-      } catch {
-        case _: ParseException =>
-          throw QueryCompilationErrors.invalidViewText(viewText, metadata.qualifiedName)
-      }
     }
     val schemaMode = metadata.viewSchemaMode
     if (schemaMode == SchemaEvolution) {

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -3464,14 +3464,6 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
       messageParameters = Map("errorMessage" -> errorMessage))
   }
 
-  def invalidViewText(viewText: String, viewName: String): Throwable = {
-    new AnalysisException(
-      errorClass = "INVALID_VIEW_TEXT",
-      messageParameters = Map(
-        "viewText" -> viewText,
-        "viewName" -> toSQLId(viewName)))
-  }
-
   def invalidTimeTravelSpecError(): Throwable = {
     new AnalysisException(
       errorClass = "INVALID_TIME_TRAVEL_SPEC",

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/SqlScriptingParserSuite.scala

@@ -48,7 +48,7 @@ class SqlScriptingParserSuite extends SparkFunSuite with SQLHelper {
     }
     assert(e.getErrorClass === "PARSE_SYNTAX_ERROR")
     assert(e.getMessage.contains("Syntax error"))
-    assert(e.getMessage.contains("SELECT 1 SELECT 1"))
+    assert(e.getMessage.contains("SELECT"))
   }
 
   test("multi select") {

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

@@ -724,9 +724,15 @@ class PersistedViewTestSuite extends SQLViewTestSuite with SharedSparkSession {
         exception = intercept[AnalysisException] {
           sql("SELECT * FROM v")
         },
-        errorClass = "INVALID_VIEW_TEXT",
-        parameters = Map(
-          "viewText" -> "DROP VIEW v", "viewName" -> tableIdentifier("v").quotedString)
+        errorClass = "PARSE_SYNTAX_ERROR",
+        parameters = Map("error" -> "'DROP'", "hint" -> ""),
+        context = ExpectedContext(
+          objectType = "VIEW",
+          objectName = "spark_catalog.default.v",
+          startIndex = 14,
+          stopIndex = 14,
+          fragment = "v"
+        )
       )
     }
   }