[SPARK-53507][CONNECT] Add breaking change info to errors

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

Adds breaking change metadata to error messages

Each breaking change includes a migration message explaining how the user should update their code. It also can include a spark config value which can be used to mitigate the breaking change.

The migration message is concatenated to the error message. In Scala, we also include the breaking change info in the structured error message, when the STANDARD error format is used.

We also include breaking change info in pyspark errors.

### Why are the changes needed?

By tagging breaking changes with metadata and a spark config flag, we can build tools to automatically retry spark jobs with the breaking change disabled.

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

This PR only adds a framework for creating breaking change errors, but does not define any breaking change errors yet. It adds new methods, for example `getBreakingChangeInfo` on `SparkThrowable`. For existing errors, this function will return `None`.

### How was this patch tested?

Tests are added in `SparkThrowableSuite`, `test_connect_errors_conversion.py`, `test_errors.py`, and `FetchErrorDetailsHandlerSuite`.

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

No

Closes #52256 from imarkowitz/ian/breaking-changes.

Authored-by: imarkowitz <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: imarkowitz

common/utils/src/main/java/org/apache/spark/SparkThrowable.java

@@ -60,6 +60,12 @@ default boolean isInternalError() {
     return SparkThrowableHelper.isInternalError(this.getCondition());
   }
 
+  // If null, the error message is not for a breaking change
+  default BreakingChangeInfo getBreakingChangeInfo() {
+    return SparkThrowableHelper.getBreakingChangeInfo(
+        this.getCondition()).getOrElse(() -> null);
+  }
+
   default Map<String, String> getMessageParameters() {
     return new HashMap<>();
   }

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

@@ -75,6 +75,22 @@ class ErrorClassesJsonReader(jsonFileURLs: Seq[URL]) {
     matches.map(m => m.stripSuffix(">").stripPrefix("<"))
   }
 
+  def getBreakingChangeInfo(errorClass: String): Option[BreakingChangeInfo] = {
+    val errorClasses = errorClass.split('.')
+    errorClasses match {
+      case Array(mainClass) =>
+        errorInfoMap.get(mainClass).flatMap(_.breakingChangeInfo)
+      case Array(mainClass, subClass) =>
+        errorInfoMap.get(mainClass).flatMap{
+          errorInfo =>
+            errorInfo.subClass.flatMap(_.get(subClass))
+              .flatMap(_.breakingChangeInfo)
+              .orElse(errorInfo.breakingChangeInfo)
+        }
+      case _ => None
+    }
+  }
+
   def getMessageTemplate(errorClass: String): String = {
     val errorClasses = errorClass.split("\\.")
     assert(errorClasses.length == 1 || errorClasses.length == 2)
@@ -128,7 +144,7 @@ private object ErrorClassesJsonReader {
     val map = mapper.readValue(url, new TypeReference[Map[String, ErrorInfo]]() {})
     val errorClassWithDots = map.collectFirst {
       case (errorClass, _) if errorClass.contains('.') => errorClass
-      case (_, ErrorInfo(_, Some(map), _)) if map.keys.exists(_.contains('.')) =>
+      case (_, ErrorInfo(_, Some(map), _, _)) if map.keys.exists(_.contains('.')) =>
         map.keys.collectFirst { case s if s.contains('.') => s }.get
     }
     if (errorClassWithDots.isEmpty) {
@@ -147,28 +163,59 @@ private object ErrorClassesJsonReader {
  * @param subClass SubClass associated with this class.
  * @param message Message format with optional placeholders (e.g. &lt;parm&gt;).
  *                The error message is constructed by concatenating the lines with newlines.
+ * @param breakingChangeInfo Additional metadata if the error is due to a breaking change.
  */
 private case class ErrorInfo(
     message: Seq[String],
     subClass: Option[Map[String, ErrorSubInfo]],
-    sqlState: Option[String]) {
+    sqlState: Option[String],
+    breakingChangeInfo: Option[BreakingChangeInfo] = None) {
   // For compatibility with multi-line error messages
   @JsonIgnore
-  val messageTemplate: String = message.mkString("\n")
+  val messageTemplate: String = message.mkString("\n") +
+    breakingChangeInfo.map(_.migrationMessage.mkString(" ", "\n", "")).getOrElse("")
 }
 
 /**
  * Information associated with an error subclass.
  *
  * @param message Message format with optional placeholders (e.g. &lt;parm&gt;).
  *                The error message is constructed by concatenating the lines with newlines.
+ * @param breakingChangeInfo Additional metadata if the error is due to a breaking change.
  */
-private case class ErrorSubInfo(message: Seq[String]) {
+private case class ErrorSubInfo(
+    message: Seq[String],
+    breakingChangeInfo: Option[BreakingChangeInfo] = None) {
   // For compatibility with multi-line error messages
   @JsonIgnore
-  val messageTemplate: String = message.mkString("\n")
+  val messageTemplate: String = message.mkString("\n") +
+      breakingChangeInfo.map(_.migrationMessage.mkString(" ", "\n", "")).getOrElse("")
 }
 
+/**
+ * Additional information if the error was caused by a breaking change.
+ *
+ * @param migrationMessage A message explaining how the user can migrate their job to work
+ *                         with the breaking change.
+ * @param mitigationConfig A spark config flag that can be used to mitigate the
+ *                              breaking change.
+ * @param needsAudit If true, the breaking change should be inspected manually.
+ *                       If false, the spark job should be retried by setting the
+ *                       mitigationConfig.
+ */
+case class BreakingChangeInfo(
+    migrationMessage: Seq[String],
+    mitigationConfig: Option[MitigationConfig] = None,
+    needsAudit: Boolean = true
+)
+
+/**
+ * A spark config flag that can be used to mitigate a breaking change.
+ * @param key The spark config key.
+ * @param value The spark config value that mitigates the breaking change.
+ */
+case class MitigationConfig(key: String, value: String)
+
 /**
  * Information associated with an error state / SQLSTATE.
  *

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

@@ -73,6 +73,14 @@ private[spark] object SparkThrowableHelper {
     errorReader.getMessageParameters(errorClass)
   }
 
+  def getBreakingChangeInfo(errorClass: String): Option[BreakingChangeInfo] = {
+    if (errorClass == null) {
+      None
+    } else {
+      errorReader.getBreakingChangeInfo(errorClass)
+    }
+  }
+
   def isInternalError(errorClass: String): Boolean = {
     errorClass != null && errorClass.startsWith("INTERNAL_ERROR")
   }
@@ -99,6 +107,19 @@ private[spark] object SparkThrowableHelper {
           g.writeStringField("errorClass", errorClass)
           if (format == STANDARD) {
             g.writeStringField("messageTemplate", errorReader.getMessageTemplate(errorClass))
+            errorReader.getBreakingChangeInfo(errorClass).foreach { breakingChangeInfo =>
+              g.writeObjectFieldStart("breakingChangeInfo")
+              g.writeStringField("migrationMessage",
+                  breakingChangeInfo.migrationMessage.mkString("\n"))
+              breakingChangeInfo.mitigationConfig.foreach { mitigationConfig =>
+                g.writeObjectFieldStart("mitigationConfig")
+                g.writeStringField("key", mitigationConfig.key)
+                g.writeStringField("value", mitigationConfig.value)
+                g.writeEndObject()
+              }
+              g.writeBooleanField("needsAudit", breakingChangeInfo.needsAudit)
+              g.writeEndObject()
+            }
           }
           val sqlState = e.getSqlState
           if (sqlState != null) g.writeStringField("sqlState", sqlState)

core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala

@@ -504,6 +504,90 @@ class SparkThrowableSuite extends SparkFunSuite {
     }
   }
 
+  test("breaking changes info") {
+    assert(SparkThrowableHelper.getBreakingChangeInfo(null).isEmpty)
+
+    val nonBreakingChangeError = new SparkException(
+      errorClass = "CANNOT_PARSE_DECIMAL",
+      messageParameters = Map.empty[String, String],
+      cause = null)
+    assert(nonBreakingChangeError.getBreakingChangeInfo == null)
+
+    withTempDir { dir =>
+      val json = new File(dir, "errors.json")
+      Files.writeString(
+        json.toPath,
+        """
+          |{
+          |  "TEST_ERROR": {
+          |    "message": [
+          |      "Error message 1 with <param1>."
+          |    ],
+          |    "breakingChangeInfo": {
+          |      "migrationMessage": [
+          |        "Migration message with <param2>."
+          |      ],
+          |      "mitigationConfig": {
+          |        "key": "config.key1",
+          |        "value": "config.value1"
+          |      },
+          |      "needsAudit": false
+          |    }
+          |  },
+          |  "TEST_ERROR_WITH_SUBCLASS": {
+          |    "message": [
+          |      "Error message 2 with <param1>."
+          |    ],
+          |    "subClass": {
+          |      "SUBCLASS": {
+          |        "message": [
+          |          "Subclass message with <param2>."
+          |        ],
+          |        "breakingChangeInfo": {
+          |          "migrationMessage": [
+          |            "Subclass migration message with <param3>."
+          |          ],
+          |          "mitigationConfig": {
+          |            "key": "config.key2",
+          |            "value": "config.value2"
+          |          },
+          |          "needsAudit": true
+          |        }
+          |      }
+          |    }
+          |  }
+          |}
+          |""".stripMargin,
+        StandardCharsets.UTF_8)
+
+      val error1Params = Map("param1" -> "value1", "param2" -> "value2")
+      val error2Params = Map("param1" -> "value1", "param2" -> "value2", "param3" -> "value3")
+
+      val reader =
+        new ErrorClassesJsonReader(Seq(errorJsonFilePath.toUri.toURL, json.toURI.toURL))
+      val errorMessage = reader.getErrorMessage("TEST_ERROR", error1Params)
+      assert(errorMessage == "Error message 1 with value1. Migration message with value2.")
+      val breakingChangeInfo = reader.getBreakingChangeInfo("TEST_ERROR")
+      assert(
+        breakingChangeInfo.contains(
+          BreakingChangeInfo(
+            Seq("Migration message with <param2>."),
+            Some(MitigationConfig("config.key1", "config.value1")),
+            needsAudit = false)))
+      val errorMessage2 =
+        reader.getErrorMessage("TEST_ERROR_WITH_SUBCLASS.SUBCLASS", error2Params)
+      assert(
+        errorMessage2 == "Error message 2 with value1. Subclass message with value2." +
+          " Subclass migration message with value3.")
+      val breakingChangeInfo2 = reader.getBreakingChangeInfo("TEST_ERROR_WITH_SUBCLASS.SUBCLASS")
+      assert(
+        breakingChangeInfo2.contains(
+          BreakingChangeInfo(
+            Seq("Subclass migration message with <param3>."),
+            Some(MitigationConfig("config.key2", "config.value2")))))
+    }
+  }
+
   test("detect unused message parameters") {
     checkError(
       exception = intercept[SparkException] {

python/pyspark/errors/exceptions/base.py

@@ -17,7 +17,7 @@
 import warnings
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import Dict, Optional, TypeVar, cast, Iterable, TYPE_CHECKING, List
+from typing import Any, Dict, Optional, TypeVar, cast, Iterable, TYPE_CHECKING, List
 
 from pyspark.errors.exceptions.tblib import Traceback
 from pyspark.errors.utils import ErrorClassesReader
@@ -138,6 +138,23 @@ def getMessage(self) -> str:
         """
         return f"[{self.getCondition()}] {self._message}"
 
+    def getBreakingChangeInfo(self) -> Optional[Dict[str, Any]]:
+        """
+        Returns the breaking change info for an error, or None.
+
+        Breaking change info is a dict with two fields:
+
+        migration_message: list of str
+            A message explaining how the user can migrate their job to work
+                with the breaking change.
+
+        mitigation_config:
+            A dict with key: str and value: str fields.
+            A spark config flag that can be used to mitigate the
+                breaking change.
+        """
+        return self._error_reader.get_breaking_change_info(self._errorClass)
+
     def getQueryContext(self) -> List["QueryContext"]:
         """
         Returns :class:`QueryContext`.

python/pyspark/errors/exceptions/connect.py

@@ -17,7 +17,7 @@
 import grpc
 import json
 from grpc import StatusCode
-from typing import Dict, List, Optional, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
 
 from pyspark.errors.exceptions.base import (
     AnalysisException as BaseAnalysisException,
@@ -95,6 +95,7 @@ def _convert_exception(
         display_server_stacktrace = display_server_stacktrace if stacktrace else False
 
     contexts = None
+    breaking_change_info = None
     if resp and resp.HasField("root_error_idx"):
         root_error = resp.errors[resp.root_error_idx]
         if hasattr(root_error, "spark_throwable"):
@@ -105,6 +106,20 @@ def _convert_exception(
                 else DataFrameQueryContext(c)
                 for c in root_error.spark_throwable.query_contexts
             ]
+            # Extract breaking change info if present
+            if hasattr(
+                root_error.spark_throwable, "breaking_change_info"
+            ) and root_error.spark_throwable.HasField("breaking_change_info"):
+                bci = root_error.spark_throwable.breaking_change_info
+                breaking_change_info = {
+                    "migration_message": list(bci.migration_message),
+                    "needs_audit": bci.needs_audit if bci.HasField("needs_audit") else True,
+                }
+                if bci.HasField("mitigation_config"):
+                    breaking_change_info["mitigation_config"] = {
+                        "key": bci.mitigation_config.key,
+                        "value": bci.mitigation_config.value,
+                    }
 
     if "org.apache.spark.api.python.PythonException" in classes:
         return PythonException(
@@ -134,6 +149,7 @@ def _convert_exception(
                 display_server_stacktrace=display_server_stacktrace,
                 contexts=contexts,
                 grpc_status_code=grpc_status_code,
+                breaking_change_info=breaking_change_info,
             )
 
     # Return UnknownException if there is no matched exception class
@@ -147,6 +163,7 @@ def _convert_exception(
         display_server_stacktrace=display_server_stacktrace,
         contexts=contexts,
         grpc_status_code=grpc_status_code,
+        breaking_change_info=breaking_change_info,
     )
 
 
@@ -193,6 +210,7 @@ def __init__(
         display_server_stacktrace: bool = False,
         contexts: Optional[List[BaseQueryContext]] = None,
         grpc_status_code: grpc.StatusCode = StatusCode.UNKNOWN,
+        breaking_change_info: Optional[Dict[str, Any]] = None,
     ) -> None:
         if contexts is None:
             contexts = []
@@ -221,6 +239,7 @@ def __init__(
         self._display_stacktrace: bool = display_server_stacktrace
         self._contexts: List[BaseQueryContext] = contexts
         self._grpc_status_code = grpc_status_code
+        self._breaking_change_info: Optional[Dict[str, Any]] = breaking_change_info
         self._log_exception()
 
     def getSqlState(self) -> Optional[str]:
@@ -241,6 +260,15 @@ def getMessage(self) -> str:
     def getGrpcStatusCode(self) -> grpc.StatusCode:
         return self._grpc_status_code
 
+    def getBreakingChangeInfo(self) -> Optional[Dict[str, Any]]:
+        """
+        Returns the breaking change info for an error, or None.
+
+        For Spark Connect exceptions, this returns the breaking change info
+        received from the server, rather than looking it up from local error files.
+        """
+        return self._breaking_change_info
+
     def __str__(self) -> str:
         return self.getMessage()
 
@@ -263,6 +291,7 @@ def __init__(
         display_server_stacktrace: bool = False,
         contexts: Optional[List[BaseQueryContext]] = None,
         grpc_status_code: grpc.StatusCode = StatusCode.UNKNOWN,
+        breaking_change_info: Optional[Dict[str, Any]] = None,
     ) -> None:
         super().__init__(
             message=message,
@@ -274,6 +303,7 @@ def __init__(
             display_server_stacktrace=display_server_stacktrace,
             contexts=contexts,
             grpc_status_code=grpc_status_code,
+            breaking_change_info=breaking_change_info,
         )