Exception Handling

/**
 * Main exception class thrown by Spark with structured error information
 * @param message Human-readable error message
 * @param cause Root cause exception (optional)
 * @param errorClass Standardized error class identifier (optional)
 * @param messageParameters Template parameters for localized messages
 * @param context Query context information for debugging
 */
class SparkException(
  message: String,
  cause: Throwable,
  errorClass: Option[String],
  messageParameters: Map[String, String],
  context: Array[QueryContext] = Array.empty
) extends Exception(message, cause) with SparkThrowable

object SparkException {
  /** Create internal error exception with message */
  def internalError(msg: String): SparkException
  
  /** Create internal error exception with message and cause */
  def internalError(msg: String, cause: Throwable): SparkException
  
  /** Create internal error exception with context and summary */
  def internalError(
    msg: String, 
    context: Array[QueryContext], 
    summary: String,
    category: Option[String]
  ): SparkException
}

import org.apache.spark.{SparkException, QueryContext}

// Basic exception creation
val basicError = new SparkException("Operation failed")

// Exception with cause
val errorWithCause = new SparkException("Database connection failed", sqlException)

// Exception with error class and parameters
val structuredError = new SparkException(
  errorClass = "INVALID_PARAMETER_VALUE",
  messageParameters = Map("parameter" -> "timeout", "value" -> "-1"),
  cause = null
)

// Internal error creation
val internalError = SparkException.internalError("Unexpected state in executor")

// Handling exceptions
try {
  // Some Spark operation
} catch {
  case e: SparkException =>
    println(s"Error class: ${e.getErrorClass}")
    println(s"Parameters: ${e.getMessageParameters}") 
    if (e.isInternalError) {
      logger.error("Internal Spark error occurred", e)
    }
}

/**
 * Interface mixed into Throwables thrown from Spark
 * Provides standardized access to error classification and context
 */
public interface SparkThrowable {
  /** Returns succinct, human-readable error category identifier */
  String getErrorClass();
  
  /** Returns portable error identifier across SQL engines (SQLSTATE) */
  default String getSqlState() {
    return SparkThrowableHelper.getSqlState(this.getErrorClass());
  }
  
  /** Returns true if this error is an internal Spark error */
  default boolean isInternalError() {
    return SparkThrowableHelper.isInternalError(this.getErrorClass());
  }
  
  /** Returns message template parameters for localization */
  default Map<String, String> getMessageParameters() {
    return new HashMap<>();
  }
  
  /** Returns query context information for debugging */
  default QueryContext[] getQueryContext() { 
    return new QueryContext[0]; 
  }
}

/**
 * Query context information for SparkThrowable exceptions
 * Helps users understand where errors occur while executing queries
 */
public interface QueryContext {
  /** 
   * The object type of the query which throws the exception
   * Empty string for main query, otherwise object type in upper case (e.g., "VIEW")
   */
  String objectType();
  
  /** 
   * The object name of the query which throws the exception  
   * Empty string for main query, otherwise the object name (e.g., view name "V1")
   */
  String objectName();
  
  /** The starting index in the query text (0-based) */
  int startIndex();
  
  /** The stopping index in the query text (0-based) */
  int stopIndex();
  
  /** The corresponding fragment of the query which throws the exception */
  String fragment();
}

/** Exception for driver process execution failures */
private[spark] class SparkDriverExecutionException(cause: Throwable) 
  extends SparkException("Execution error", cause)

/** Exception for child process exit codes in PySpark */  
private[spark] case class SparkUserAppException(exitCode: Int)
  extends SparkException(s"User application exited with $exitCode")

/** Exception for dead executor access attempts */
private[spark] case class ExecutorDeadException(message: String)
  extends SparkException(
    errorClass = "INTERNAL_ERROR_NETWORK",
    messageParameters = Map("message" -> message), 
    cause = null
  )

/** Exception for version upgrade compatibility issues */
private[spark] class SparkUpgradeException private(
  message: String,
  cause: Option[Throwable], 
  errorClass: Option[String],
  messageParameters: Map[String, String]
) extends RuntimeException with SparkThrowable

/** Arithmetic exception with error classification */
private[spark] class SparkArithmeticException private(
  message: String,
  errorClass: Option[String],
  messageParameters: Map[String, String],
  context: Array[QueryContext]  
) extends ArithmeticException(message) with SparkThrowable

/** Other specialized exceptions for specific domains */
private[spark] class SparkUnsupportedOperationException // For unsupported operations
private[spark] class SparkClassNotFoundException // For missing classes
private[spark] class SparkConcurrentModificationException // For concurrency issues
private[spark] class SparkDateTimeException // For date/time errors
private[spark] class SparkFileNotFoundException // For missing files
private[spark] class SparkNumberFormatException // For number parsing errors
private[spark] class SparkIllegalArgumentException // For invalid arguments
private[spark] class SparkRuntimeException // For runtime errors
private[spark] class SparkNoSuchElementException // For missing elements
private[spark] class SparkSecurityException // For security violations
private[spark] class SparkArrayIndexOutOfBoundsException // For array access errors
private[spark] class SparkSQLException // For SQL-related errors
private[spark] class SparkSQLFeatureNotSupportedException // For unsupported SQL features

// Prefer error classes over plain messages for new code
val standardError = new SparkException(
  errorClass = "INVALID_PARAMETER_VALUE",
  messageParameters = Map("parameter" -> "partitions", "value" -> "0"),
  cause = null
)

// Use internal error factory methods for internal failures
val internalError = SparkException.internalError(
  "Unexpected null value in partition metadata"
)

// Include query context for user-facing errors
val contextualError = new SparkException(
  errorClass = "SYNTAX_ERROR", 
  messageParameters = Map("message" -> "Invalid column reference"),
  cause = null,
  context = Array(queryContext)
)

try {
  // Spark operations
} catch {
  case e: SparkException if e.isInternalError =>
    // Log internal errors for debugging
    logger.error(s"Internal Spark error: ${e.getErrorClass}", e)
    throw e
    
  case e: SparkException =>
    // Handle user errors gracefully
    val errorInfo = s"Error: ${e.getErrorClass} - ${e.getMessage}"
    logger.warn(errorInfo)
    // Potentially recoverable - retry or provide user feedback
    
  case e: SparkThrowable =>
    // Handle other Spark throwables
    if (e.getSqlState != null) {
      // SQL-compatible error handling
      handleSqlError(e.getSqlState, e.getMessage)
    }
}