[SPARK-54883][MINOR] Clean up error messages for CLI and add new error mode DEBUG

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

Two changes in this pull request:
- Suppress stack trace information fro spark-sql CLI with PRETTY error format, unless the error is internal. Prior to this stack trace was suppressed only for AnalysisException
- Introduce a new DEBUG error format which matches PRETTY + always dumping the stack (if available).

### Why are the changes needed?

This makes the experience of using CLI much more consistent without losing capability. It also clearly seperates "consumer" output, from developer output.

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

Yes, in CLI error message output is by default more consistent.

### How was this patch tested?

Added new tests

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

Claude

Closes #53659 from srielau/spark-sql.

Authored-by: Serge Rielau <serge@rielau.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: srielau

common/utils/src/main/scala/org/apache/spark/SparkThrowableHelper.scala

@@ -23,7 +23,7 @@ import org.apache.spark.util.JsonUtils.toJsonString
 import org.apache.spark.util.SparkClassUtils
 
 private[spark] object ErrorMessageFormat extends Enumeration {
-  val PRETTY, MINIMAL, STANDARD = Value
+  val PRETTY, MINIMAL, STANDARD, DEBUG = Value
 }
 
 /**
@@ -113,7 +113,7 @@ private[spark] object SparkThrowableHelper {
   def getMessage(e: SparkThrowable with Throwable, format: ErrorMessageFormat.Value): String = {
     import ErrorMessageFormat._
     format match {
-      case PRETTY => e.getMessage
+      case PRETTY | DEBUG => e.getMessage
       case MINIMAL | STANDARD if e.getCondition == null =>
         toJsonString { generator =>
           val g = generator.useDefaultPrettyPrinter()

project/MimaExcludes.scala

@@ -35,6 +35,8 @@ object MimaExcludes {
 
   // Exclude rules for 4.2.x from 4.1.0
   lazy val v42excludes = v41excludes ++ Seq(
+    // Add DEBUG format to ErrorMessageFormat enum
+    ProblemFilters.exclude[Problem]("org.apache.spark.ErrorMessageFormat*")
   )
 
   // Exclude rules for 4.1.x from 4.0.0

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -6193,7 +6193,9 @@ object SQLConf {
 
   val ERROR_MESSAGE_FORMAT = buildConf("spark.sql.error.messageFormat")
     .doc("When PRETTY, the error message consists of textual representation of error class, " +
-      "message and query context. The MINIMAL and STANDARD formats are pretty JSON formats where " +
+      "message and query context. Stack traces are only shown for internal errors " +
+      "(SQLSTATE XX***). When DEBUG, the output is the same as PRETTY but stack traces are " +
+      "always included. The MINIMAL and STANDARD formats are pretty JSON formats where " +
       "STANDARD includes an additional JSON field `message`. This configuration property " +
       "influences on error messages of Thrift Server and SQL CLI while running queries.")
     .version("3.4.0")

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala

@@ -40,7 +40,6 @@ import org.apache.spark.{ErrorMessageFormat, SparkConf, SparkThrowable, SparkThr
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys._
-import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
 import org.apache.spark.sql.catalyst.util.SQLKeywordUtils
 import org.apache.spark.sql.hive.client.HiveClientImpl
@@ -452,9 +451,24 @@ private[hive] class SparkSQLCLIDriver extends CliDriver with Logging {
                 case _ => t.getMessage
               }
               err.println(msg)
-              if (format == ErrorMessageFormat.PRETTY &&
-                !sessionState.getIsSilent &&
-                (!t.isInstanceOf[AnalysisException] || t.getCause != null)) {
+              // Print stack traces based on format and error type:
+              // - DEBUG format: Always print stack traces (for debugging)
+              // - PRETTY format: Only for internal errors (SQLSTATE XX***)
+              // - MINIMAL/STANDARD formats: Never print stack traces (JSON only)
+              val shouldPrintStackTrace = format match {
+                case ErrorMessageFormat.DEBUG => true // Always print in DEBUG mode
+                case ErrorMessageFormat.PRETTY =>
+                  // In PRETTY mode, only print for internal errors
+                  t match {
+                    case st: SparkThrowable =>
+                      val sqlState = st.getSqlState
+                      // Print if: internal error (XX***) OR no SQLSTATE
+                      sqlState == null || sqlState.startsWith("XX")
+                    case _ => true // Non-SparkThrowable exceptions always get stack traces
+                  }
+                case _ => false // MINIMAL and STANDARD never print stack traces
+              }
+              if (shouldPrintStackTrace && !sessionState.getIsSilent) {
                 t.printStackTrace(err)
               }
               driver.close()
@@ -688,4 +702,3 @@ private[hive] class SparkSQLCLIDriver extends CliDriver with Logging {
     ret
   }
 }
-

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala

@@ -554,22 +554,26 @@ class CliSuite extends SparkFunSuite {
     )
   }
 
-  testRetry("SparkException with root cause will be printStacktrace") {
-    // If it is not in silent mode, will print the stacktrace
+  testRetry("DEBUG format prints stack traces for all errors") {
+    // In DEBUG format with silent=false, stack traces are always printed
     runCliWithin(
       1.minute,
-      extraArgs = Seq("--hiveconf", "hive.session.silent=false",
+      extraArgs = Seq(
+        "--hiveconf", "hive.session.silent=false",
+        "--conf", s"${SQLConf.ERROR_MESSAGE_FORMAT.key}=${ErrorMessageFormat.DEBUG}",
         "-e", "select from_json('a', 'a INT', map('mode', 'FAILFAST'));"),
       errorResponses = Seq("JsonParseException"))(
-      ("", "SparkException: [MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION]"),
+      ("", "[MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION]"),
       ("", "JsonParseException: Unrecognized token 'a'"))
-    // If it is in silent mode, will print the error message only
+    // In DEBUG format with silent=true, stack traces are suppressed
     runCliWithin(
       1.minute,
-      extraArgs = Seq("--conf", "spark.hive.session.silent=true",
+      extraArgs = Seq(
+        "--conf", "spark.hive.session.silent=true",
+        "--conf", s"${SQLConf.ERROR_MESSAGE_FORMAT.key}=${ErrorMessageFormat.DEBUG}",
         "-e", "select from_json('a', 'a INT', map('mode', 'FAILFAST'));"),
-      errorResponses = Seq("SparkException"))(
-      ("", "SparkException: [MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION]"))
+      errorResponses = Seq("MALFORMED_RECORD_IN_PARSING"))(
+      ("", "[MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION]"))
   }
 
   test("SPARK-30808: use Java 8 time API in Thrift SQL CLI by default") {
@@ -714,6 +718,8 @@ class CliSuite extends SparkFunSuite {
           "-e", "select 1 / 0"),
         errorResponses = Seq("DIVIDE_BY_ZERO"))(expected.toImmutableArraySeq: _*)
     }
+    // DIVIDE_BY_ZERO has SQLSTATE 22012 (not XX***), so it's a user error
+    // It also has no cause, so stack traces should NOT be shown in PRETTY mode
     check(
       format = ErrorMessageFormat.PRETTY,
       errorMessage =
@@ -725,6 +731,26 @@ class CliSuite extends SparkFunSuite {
       silent = true)
     check(
       format = ErrorMessageFormat.PRETTY,
+      errorMessage =
+        """[DIVIDE_BY_ZERO] Division by zero. Use `try_divide` to tolerate divisor being 0 and return NULL instead. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error.
+          |== SQL (line 1, position 8) ==
+          |select 1 / 0
+          |       ^^^^^
+          |""".stripMargin,
+      silent = false)
+    // DEBUG format should always show stack traces, even for user errors
+    // Silent mode still suppresses stack traces
+    check(
+      format = ErrorMessageFormat.DEBUG,
+      errorMessage =
+        """[DIVIDE_BY_ZERO] Division by zero. Use `try_divide` to tolerate divisor being 0 and return NULL instead. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error.
+          |== SQL (line 1, position 8) ==
+          |select 1 / 0
+          |       ^^^^^
+          |""".stripMargin,
+      silent = true)
+    check(
+      format = ErrorMessageFormat.DEBUG,
       errorMessage =
         """[DIVIDE_BY_ZERO] Division by zero. Use `try_divide` to tolerate divisor being 0 and return NULL instead. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error.
           |== SQL (line 1, position 8) ==