native/common/src/error.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use arrow::error::ArrowError;
use datafusion::common::DataFusionError;
use std::sync::Arc;

#[derive(thiserror::Error, Debug, Clone)]
pub enum SparkError {
    #[error(
        "[MALFORMED_VARIANT] Variant binary is malformed. Please check the data source is valid."
    )]
    MalformedVariant,

    // This list was generated from the Spark code. Many of the exceptions are not yet used by Comet
    #[error("[CAST_INVALID_INPUT] The value '{value}' of the type \"{from_type}\" cannot be cast to \"{to_type}\" \
        because it is malformed. Correct the value as per the syntax, or change its target type. \
        Use `try_cast` to tolerate malformed input and return NULL instead. If necessary \
        set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    CastInvalidValue {
        value: String,
        from_type: String,
        to_type: String,
    },

    /// Like CastInvalidValue but maps to SparkDateTimeException instead of SparkNumberFormatException.
    /// Used for string → timestamp/date cast failures.
    #[error("[CAST_INVALID_INPUT] The value '{value}' of the type \"{from_type}\" cannot be cast to \"{to_type}\" \
        because it is malformed. Correct the value as per the syntax, or change its target type. \
        Use `try_cast` to tolerate malformed input and return NULL instead. If necessary \
        set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    InvalidInputInCastToDatetime {
        value: String,
        from_type: String,
        to_type: String,
    },

    #[error("[NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION] {value} cannot be represented as Decimal({precision}, {scale}). If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error, and return NULL instead.")]
    NumericValueOutOfRange {
        value: String,
        precision: u8,
        scale: i8,
    },

    #[error("[NUMERIC_OUT_OF_SUPPORTED_RANGE] The value {value} cannot be interpreted as a numeric since it has more than 38 digits.")]
    NumericOutOfRange { value: String },

    #[error("[CAST_OVERFLOW] The value {value} of the type \"{from_type}\" cannot be cast to \"{to_type}\" \
        due to an overflow. Use `try_cast` to tolerate overflow and return NULL instead. If necessary \
        set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    CastOverFlow {
        value: String,
        from_type: String,
        to_type: String,
    },

    #[error("[CANNOT_PARSE_DECIMAL] Cannot parse decimal.")]
    CannotParseDecimal,

    #[error("[ARITHMETIC_OVERFLOW] {from_type} overflow.{suggestion} If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.",
        suggestion = if function_name.is_empty() { String::new() } else {
            format!(" Use '{function_name}' to tolerate overflow and return NULL instead.")
        })]
    ArithmeticOverflow {
        from_type: String,
        function_name: String,
    },

    #[error("[ARITHMETIC_OVERFLOW] Overflow in integral divide. Use 'try_divide' to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    IntegralDivideOverflow,

    #[error("[ARITHMETIC_OVERFLOW] Overflow in sum of decimals. Use `try_{function_name}` to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    DecimalSumOverflow { function_name: String },

    #[error("[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.")]
    DivideByZero,

    #[error("[REMAINDER_BY_ZERO] Remainder by zero. Use `try_mod` to tolerate divisor being 0 and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    RemainderByZero,

    #[error("[INTERVAL_DIVIDED_BY_ZERO] Divide by zero in interval arithmetic.")]
    IntervalDividedByZero,

    #[error("[BINARY_ARITHMETIC_OVERFLOW] {value1} {symbol} {value2} caused overflow. Use `{function_name}` to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    BinaryArithmeticOverflow {
        value1: String,
        symbol: String,
        value2: String,
        function_name: String,
    },

    #[error("[INTERVAL_ARITHMETIC_OVERFLOW.WITH_SUGGESTION] Interval arithmetic overflow. Use `{function_name}` to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    IntervalArithmeticOverflowWithSuggestion { function_name: String },

    #[error("[INTERVAL_ARITHMETIC_OVERFLOW.WITHOUT_SUGGESTION] Interval arithmetic overflow. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    IntervalArithmeticOverflowWithoutSuggestion,

    #[error("[DATETIME_OVERFLOW] Datetime arithmetic overflow.")]
    DatetimeOverflow,

    #[error("[ILLEGAL_DAY_OF_WEEK] Illegal input for day of week: {input}.")]
    IllegalDayOfWeek { input: String },

    #[error("[DATETIME_FIELD_OUT_OF_BOUNDS] {range_message}. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    DatetimeFieldOutOfBounds { range_message: String },

    #[error("[INVALID_ARRAY_INDEX] The index {index_value} is out of bounds. The array has {array_size} elements. Use the SQL function get() to tolerate accessing element at invalid index and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    InvalidArrayIndex { index_value: i32, array_size: i32 },

    #[error("[INVALID_ARRAY_INDEX_IN_ELEMENT_AT] The index {index_value} is out of bounds. The array has {array_size} elements. Use try_element_at to tolerate accessing element at invalid index and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    InvalidElementAtIndex { index_value: i32, array_size: i32 },

    #[error("[INVALID_BITMAP_POSITION] The bit position {bit_position} is out of bounds. The bitmap has {bitmap_num_bytes} bytes ({bitmap_num_bits} bits).")]
    InvalidBitmapPosition {
        bit_position: i64,
        bitmap_num_bytes: i64,
        bitmap_num_bits: i64,
    },

    #[error("[INVALID_INDEX_OF_ZERO] The index 0 is invalid. An index shall be either < 0 or > 0 (the first element has index 1).")]
    InvalidIndexOfZero,

    #[error("[DUPLICATED_MAP_KEY] Cannot create map with duplicate keys: {key}.")]
    DuplicatedMapKey { key: String },

    #[error("[NULL_MAP_KEY] Cannot use null as map key.")]
    NullMapKey,

    #[error("[MAP_KEY_VALUE_DIFF_SIZES] The key array and value array of a map must have the same length.")]
    MapKeyValueDiffSizes,

    #[error("[EXCEED_LIMIT_LENGTH] Cannot create a map with {size} elements which exceeds the limit {max_size}.")]
    ExceedMapSizeLimit { size: i32, max_size: i32 },

    /// `num_elements` is a decimal string because Spark reports the unclamped length, which can
    /// exceed i64 (e.g. sequence(Long.MinValue, Long.MaxValue, 1)). The JVM shim maps this to the
    /// version-appropriate `createArrayWithElementsExceedLimitError`, passing `function_name`
    /// through on Spark 4.x (which includes it in the message) and ignoring it on 3.x.
    #[error("[COLLECTION_SIZE_LIMIT_EXCEEDED] Can't create array with {num_elements} elements which exceeding the array size limit {max_elements}.")]
    CollectionSizeLimitExceeded {
        num_elements: String,
        max_elements: i64,
        function_name: String,
    },

    /// Step direction does not match the start/stop bounds in `sequence`. The JVM shim maps this
    /// to a plain IllegalArgumentException on Spark 3.x and SparkIllegalArgumentException
    /// (_LEGACY_ERROR_TEMP_3243) on 4.x, matching what Spark's codegen throws.
    #[error("[_LEGACY_ERROR_TEMP_3243] Illegal sequence boundaries: {start} to {stop} by {step}")]
    SequenceIllegalBoundaries {
        start: String,
        stop: String,
        step: String,
    },

    /// Sum of every row's sequence length in one Arrow batch exceeds Comet's per-batch offset
    /// ceiling (i32::MAX) or the allocator could not satisfy the reservation. Spark itself has
    /// no equivalent limit because it stores each row as its own `long[]`. Reported with a
    /// message that names `spark.comet.batchSize` as the actionable knob.
    #[error(
        "Comet's native `sequence` kernel cannot materialize a batch with {total_elements} \
        total elements: it exceeds the per-batch limit or the allocator refused the reservation. \
        Lower `spark.comet.batchSize` so fewer rows are grouped per batch."
    )]
    SequenceBatchTooLarge { total_elements: String },

    #[error("[NOT_NULL_ASSERT_VIOLATION] The field `{field_name}` cannot be null.")]
    NotNullAssertViolation { field_name: String },

    #[error("[VALUE_IS_NULL] The value of field `{field_name}` at row {row_index} is null.")]
    ValueIsNull { field_name: String, row_index: i32 },

    #[error("[CANNOT_PARSE_TIMESTAMP] Cannot parse timestamp: {message}. Try using `{suggested_func}` instead.")]
    CannotParseTimestamp {
        message: String,
        suggested_func: String,
    },

    #[error("[INVALID_FRACTION_OF_SECOND] The fraction of second {value} is invalid. Valid values are in the range [0, 60]. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    InvalidFractionOfSecond { value: f64 },

    #[error("[INVALID_UTF8_STRING] Invalid UTF-8 string: {hex_string}.")]
    InvalidUtf8String { hex_string: String },

    #[error("[UNEXPECTED_POSITIVE_VALUE] The {parameter_name} parameter must be less than or equal to 0. The actual value is {actual_value}.")]
    UnexpectedPositiveValue {
        parameter_name: String,
        actual_value: i32,
    },

    #[error("[UNEXPECTED_NEGATIVE_VALUE] The {parameter_name} parameter must be greater than or equal to 0. The actual value is {actual_value}.")]
    UnexpectedNegativeValue {
        parameter_name: String,
        actual_value: i32,
    },

    #[error("[INVALID_PARAMETER_VALUE] Invalid regex group index {group_index} in function `{function_name}`. Group count is {group_count}.")]
    InvalidRegexGroupIndex {
        function_name: String,
        group_count: i32,
        group_index: i32,
    },

    #[error("[INVALID_URL] The url is invalid: {url}. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
    InvalidUrl { url: String },

    #[error("[DATATYPE_CANNOT_ORDER] Cannot order by type: {data_type}.")]
    DatatypeCannotOrder { data_type: String },

    #[error("[SCALAR_SUBQUERY_TOO_MANY_ROWS] Scalar subquery returned more than one row.")]
    ScalarSubqueryTooManyRows,

    /// Mirrors Spark's `QueryExecutionErrors.mergeCardinalityViolationError()`, raised by
    /// `MergeRowsExec.BitmapCardinalityValidator` when a MERGE's ON condition matches a single
    /// target row against more than one source row.
    #[error("[MERGE_CARDINALITY_VIOLATION] The ON search condition of the MERGE statement matched a single row from the target table with multiple rows of the source table. This could result in the target row being operated on more than once with an update or delete operation and is not allowed.")]
    MergeCardinalityViolation,

    #[error("{message}")]
    FileNotFound { message: String },

    #[error("[_LEGACY_ERROR_TEMP_2093] Found duplicate field(s) \"{required_field_name}\": [{matched_fields}] in case-insensitive mode")]
    DuplicateFieldCaseInsensitive {
        required_field_name: String,
        matched_fields: String,
    },

    /// Multiple Parquet fields share the same field id when the read schema requested an
    /// id-based lookup. Mirrors Spark's `_LEGACY_ERROR_TEMP_2094`
    /// (`foundDuplicateFieldInFieldIdLookupModeError`).
    #[error("[_LEGACY_ERROR_TEMP_2094] Found duplicate field(s) by id: id={required_id} matches {matched_fields} in id-lookup mode")]
    DuplicateFieldByFieldId {
        required_id: i32,
        matched_fields: String,
    },

    /// The read schema carries Parquet field ids but the file carries none. Mirrors the runtime
    /// error raised in Spark's `ParquetReadSupport` when
    /// `spark.sql.parquet.fieldId.read.ignoreMissing` is false. Spark's message names no file,
    /// so `file_path` travels as a parameter for the 4.x shim's `FAILED_READ_FILE` wrapper.
    #[error("Spark read schema expects field Ids, but Parquet file schema doesn't contain any field Ids. Please remove the field ids from Spark schema or ignore missing ids by setting `spark.sql.parquet.fieldId.read.ignoreMissing = true`")]
    ParquetMissingFieldIds { file_path: String },

    /// Schema mismatch when reading a Parquet column under a requested schema
    /// that's incompatible with the physical column type. Translated by the JVM
    /// shim into Spark's `SchemaColumnConvertNotSupportedException`. The
    /// `file_path` may be empty when the rejection happens at planning time
    /// (the schema adapter doesn't carry the file path).
    #[error("Parquet column cannot be converted in file {file_path}. Column: [{column}], Expected: {spark_type}, Found: {physical_type}")]
    ParquetSchemaConvert {
        file_path: String,
        column: String,
        physical_type: String,
        spark_type: String,
    },

    /// A per-file read failure (corrupt footer/page, truncated/empty file, deleted file) raised by
    /// the native parquet reader / object_store. Classified by typed `DataFusionError` variant (no
    /// message matching) and translated by the JVM shim into Spark's `FAILED_READ_FILE`
    /// (`QueryExecutionErrors.cannotReadFilesError`). `file_path` may be empty when the underlying
    /// error doesn't carry it (only `object_store::Error::NotFound` does); the JVM side then fills
    /// it from the per-task file list.
    #[error("Encountered error while reading file {file_path}: {message}")]
    CannotReadFile { file_path: String, message: String },

    /// A native scan refused to rebase an ancient date or timestamp under the EXCEPTION rebase
    /// mode. Converted by the JVM shim with `DataSourceUtils.newRebaseExceptionInRead(format)`,
    /// which throws on a format it does not know. `format` must be "Parquet" or "Parquet INT96",
    /// so build it with [`SparkError::read_ancient_datetime`]. `column` only feeds the native
    /// message: Spark's exception has no column parameter, so the JVM drops it.
    #[error("[INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME] Reading dates before 1582-10-15 or timestamps before 1900-01-01T00:00:00Z from {format} files can be ambiguous (column {column})")]
    ReadAncientDatetime { format: String, column: String },

    #[error("ArrowError: {0}.")]
    Arrow(Arc<ArrowError>),

    #[error("InternalError: {0}.")]
    Internal(String),
}

impl SparkError {
    /// Serialize this error to JSON format for JNI transfer
    pub fn to_json(&self) -> String {
        let error_class = self.error_class().unwrap_or("");

        // Create a JSON structure with errorType, errorClass, and params
        match serde_json::to_string(&serde_json::json!({
            "errorType": self.error_type_name(),
            "errorClass": error_class,
            "params": self.params_as_json(),
        })) {
            Ok(json) => json,
            Err(e) => {
                // Fallback if serialization fails
                format!(
                    "{{\"errorType\":\"SerializationError\",\"message\":\"{}\"}}",
                    e
                )
            }
        }
    }

    /// Construct a [`SparkError::DuplicateFieldCaseInsensitive`], formatting `matched` as the
    /// bracketed, comma-joined list the JVM `ShimSparkErrorConverter` consumes (fed verbatim into
    /// the `matchedOrcFields` param). Centralizes that format so every producer stays consistent.
    pub fn duplicate_field_case_insensitive(
        required_field_name: &str,
        matched: &[&str],
    ) -> SparkError {
        SparkError::DuplicateFieldCaseInsensitive {
            required_field_name: required_field_name.to_string(),
            matched_fields: format!("[{}]", matched.join(", ")),
        }
    }

    /// Construct a [`SparkError::ReadAncientDatetime`] for `column`, with Spark's "Parquet INT96"
    /// format when the column is an INT96 timestamp and "Parquet" otherwise.
    pub fn read_ancient_datetime(column: &str, is_int96: bool) -> SparkError {
        let format = if is_int96 { "Parquet INT96" } else { "Parquet" };
        SparkError::ReadAncientDatetime {
            format: format.to_string(),
            column: column.to_string(),
        }
    }

    /// Get the error type name for JSON serialization
    pub(crate) fn error_type_name(&self) -> &'static str {
        match self {
            SparkError::MalformedVariant => "MalformedVariant",
            SparkError::CastInvalidValue { .. } => "CastInvalidValue",
            SparkError::InvalidInputInCastToDatetime { .. } => "InvalidInputInCastToDatetime",
            SparkError::NumericValueOutOfRange { .. } => "NumericValueOutOfRange",
            SparkError::NumericOutOfRange { .. } => "NumericOutOfRange",
            SparkError::CastOverFlow { .. } => "CastOverFlow",
            SparkError::CannotParseDecimal => "CannotParseDecimal",
            SparkError::ArithmeticOverflow { .. } => "ArithmeticOverflow",
            SparkError::IntegralDivideOverflow => "IntegralDivideOverflow",
            SparkError::DecimalSumOverflow { .. } => "DecimalSumOverflow",
            SparkError::DivideByZero => "DivideByZero",
            SparkError::RemainderByZero => "RemainderByZero",
            SparkError::IntervalDividedByZero => "IntervalDividedByZero",
            SparkError::BinaryArithmeticOverflow { .. } => "BinaryArithmeticOverflow",
            SparkError::IntervalArithmeticOverflowWithSuggestion { .. } => {
                "IntervalArithmeticOverflowWithSuggestion"
            }
            SparkError::IntervalArithmeticOverflowWithoutSuggestion => {
                "IntervalArithmeticOverflowWithoutSuggestion"
            }
            SparkError::DatetimeOverflow => "DatetimeOverflow",
            SparkError::IllegalDayOfWeek { .. } => "IllegalDayOfWeek",
            SparkError::DatetimeFieldOutOfBounds { .. } => "DatetimeFieldOutOfBounds",
            SparkError::InvalidArrayIndex { .. } => "InvalidArrayIndex",
            SparkError::InvalidElementAtIndex { .. } => "InvalidElementAtIndex",
            SparkError::InvalidBitmapPosition { .. } => "InvalidBitmapPosition",
            SparkError::InvalidIndexOfZero => "InvalidIndexOfZero",
            SparkError::DuplicatedMapKey { .. } => "DuplicatedMapKey",
            SparkError::NullMapKey => "NullMapKey",
            SparkError::MapKeyValueDiffSizes => "MapKeyValueDiffSizes",
            SparkError::ExceedMapSizeLimit { .. } => "ExceedMapSizeLimit",
            SparkError::CollectionSizeLimitExceeded { .. } => "CollectionSizeLimitExceeded",
            SparkError::SequenceIllegalBoundaries { .. } => "SequenceIllegalBoundaries",
            SparkError::SequenceBatchTooLarge { .. } => "SequenceBatchTooLarge",
            SparkError::NotNullAssertViolation { .. } => "NotNullAssertViolation",
            SparkError::ValueIsNull { .. } => "ValueIsNull",
            SparkError::CannotParseTimestamp { .. } => "CannotParseTimestamp",
            SparkError::InvalidFractionOfSecond { .. } => "InvalidFractionOfSecond",
            SparkError::InvalidUtf8String { .. } => "InvalidUtf8String",
            SparkError::UnexpectedPositiveValue { .. } => "UnexpectedPositiveValue",
            SparkError::UnexpectedNegativeValue { .. } => "UnexpectedNegativeValue",
            SparkError::InvalidRegexGroupIndex { .. } => "InvalidRegexGroupIndex",
            SparkError::InvalidUrl { .. } => "InvalidUrl",
            SparkError::DatatypeCannotOrder { .. } => "DatatypeCannotOrder",
            SparkError::ScalarSubqueryTooManyRows => "ScalarSubqueryTooManyRows",
            SparkError::MergeCardinalityViolation => "MergeCardinalityViolation",
            SparkError::FileNotFound { .. } => "FileNotFound",
            SparkError::DuplicateFieldCaseInsensitive { .. } => "DuplicateFieldCaseInsensitive",
            SparkError::DuplicateFieldByFieldId { .. } => "DuplicateFieldByFieldId",
            SparkError::ParquetMissingFieldIds { .. } => "ParquetMissingFieldIds",
            SparkError::ParquetSchemaConvert { .. } => "ParquetSchemaConvert",
            SparkError::CannotReadFile { .. } => "CannotReadFile",
            SparkError::ReadAncientDatetime { .. } => "ReadAncientDatetime",
            SparkError::Arrow(_) => "Arrow",
            SparkError::Internal(_) => "Internal",
        }
    }

    /// Extract parameters as JSON value
    pub(crate) fn params_as_json(&self) -> serde_json::Value {
        match self {
            SparkError::CastInvalidValue {
                value,
                from_type,
                to_type,
            } => {
                serde_json::json!({
                    "value": value,
                    "fromType": from_type,
                    "toType": to_type,
                })
            }
            SparkError::InvalidInputInCastToDatetime {
                value,
                from_type,
                to_type,
            } => {
                serde_json::json!({
                    "value": value,
                    "fromType": from_type,
                    "toType": to_type,
                })
            }
            SparkError::NumericValueOutOfRange {
                value,
                precision,
                scale,
            } => {
                serde_json::json!({
                    "value": value,
                    "precision": precision,
                    "scale": scale,
                })
            }
            SparkError::NumericOutOfRange { value } => {
                serde_json::json!({
                    "value": value,
                })
            }
            SparkError::CastOverFlow {
                value,
                from_type,
                to_type,
            } => {
                serde_json::json!({
                    "value": value,
                    "fromType": from_type,
                    "toType": to_type,
                })
            }
            SparkError::ArithmeticOverflow {
                from_type,
                function_name,
            } => {
                serde_json::json!({
                    "fromType": from_type,
                    "functionName": function_name,
                })
            }
            SparkError::DecimalSumOverflow { function_name } => {
                serde_json::json!({
                    "functionName": function_name,
                })
            }
            SparkError::BinaryArithmeticOverflow {
                value1,
                symbol,
                value2,
                function_name,
            } => {
                serde_json::json!({
                    "value1": value1,
                    "symbol": symbol,
                    "value2": value2,
                    "functionName": function_name,
                })
            }
            SparkError::IntervalArithmeticOverflowWithSuggestion { function_name } => {
                serde_json::json!({
                    "functionName": function_name,
                })
            }
            SparkError::InvalidArrayIndex {
                index_value,
                array_size,
            } => {
                serde_json::json!({
                    "indexValue": index_value,
                    "arraySize": array_size,
                })
            }
            SparkError::InvalidElementAtIndex {
                index_value,
                array_size,
            } => {
                serde_json::json!({
                    "indexValue": index_value,
                    "arraySize": array_size,
                })
            }
            SparkError::InvalidBitmapPosition {
                bit_position,
                bitmap_num_bytes,
                bitmap_num_bits,
            } => {
                serde_json::json!({
                    "bitPosition": bit_position,
                    "bitmapNumBytes": bitmap_num_bytes,
                    "bitmapNumBits": bitmap_num_bits,
                })
            }
            SparkError::DuplicatedMapKey { key } => {
                serde_json::json!({
                    "key": key,
                })
            }
            SparkError::ExceedMapSizeLimit { size, max_size } => {
                serde_json::json!({
                    "size": size,
                    "maxSize": max_size,
                })
            }
            SparkError::CollectionSizeLimitExceeded {
                num_elements,
                max_elements,
                function_name,
            } => {
                serde_json::json!({
                    "numElements": num_elements,
                    "maxElements": max_elements,
                    "functionName": function_name,
                })
            }
            SparkError::SequenceIllegalBoundaries { start, stop, step } => {
                serde_json::json!({
                    "start": start,
                    "stop": stop,
                    "step": step,
                })
            }
            SparkError::SequenceBatchTooLarge { total_elements } => {
                serde_json::json!({
                    "totalElements": total_elements,
                })
            }
            SparkError::NotNullAssertViolation { field_name } => {
                serde_json::json!({
                    "fieldName": field_name,
                })
            }
            SparkError::ValueIsNull {
                field_name,
                row_index,
            } => {
                serde_json::json!({
                    "fieldName": field_name,
                    "rowIndex": row_index,
                })
            }
            SparkError::CannotParseTimestamp {
                message,
                suggested_func,
            } => {
                serde_json::json!({
                    "message": message,
                    "suggestedFunc": suggested_func,
                })
            }
            SparkError::IllegalDayOfWeek { input } => {
                serde_json::json!({
                    "string": input,
                })
            }
            SparkError::DatetimeFieldOutOfBounds { range_message } => {
                serde_json::json!({
                    "rangeMessage": range_message,
                })
            }
            SparkError::InvalidFractionOfSecond { value } => {
                serde_json::json!({
                    "value": value,
                })
            }
            SparkError::InvalidUtf8String { hex_string } => {
                serde_json::json!({
                    "hexString": hex_string,
                })
            }
            SparkError::UnexpectedPositiveValue {
                parameter_name,
                actual_value,
            } => {
                serde_json::json!({
                    "parameterName": parameter_name,
                    "actualValue": actual_value,
                })
            }
            SparkError::UnexpectedNegativeValue {
                parameter_name,
                actual_value,
            } => {
                serde_json::json!({
                    "parameterName": parameter_name,
                    "actualValue": actual_value,
                })
            }
            SparkError::InvalidRegexGroupIndex {
                function_name,
                group_count,
                group_index,
            } => {
                serde_json::json!({
                    "functionName": function_name,
                    "groupCount": group_count,
                    "groupIndex": group_index,
                })
            }
            SparkError::InvalidUrl { url } => {
                serde_json::json!({
                    "url": url,
                })
            }
            SparkError::DatatypeCannotOrder { data_type } => {
                serde_json::json!({
                    "dataType": data_type,
                })
            }
            SparkError::FileNotFound { message } => {
                serde_json::json!({
                    "message": message,
                })
            }
            SparkError::DuplicateFieldCaseInsensitive {
                required_field_name,
                matched_fields,
            } => {
                serde_json::json!({
                    "requiredFieldName": required_field_name,
                    "matchedOrcFields": matched_fields,
                })
            }
            SparkError::DuplicateFieldByFieldId {
                required_id,
                matched_fields,
            } => {
                serde_json::json!({
                    "requiredId": required_id,
                    "matchedFields": matched_fields,
                })
            }
            SparkError::ParquetMissingFieldIds { file_path } => {
                serde_json::json!({
                    "filePath": file_path,
                })
            }
            SparkError::ParquetSchemaConvert {
                file_path,
                column,
                physical_type,
                spark_type,
            } => {
                serde_json::json!({
                    "filePath": file_path,
                    "column": column,
                    "physicalType": physical_type,
                    "sparkType": spark_type,
                })
            }
            SparkError::CannotReadFile { file_path, message } => {
                serde_json::json!({
                    "filePath": file_path,
                    "message": message,
                })
            }
            SparkError::ReadAncientDatetime { format, column } => {
                serde_json::json!({
                    "format": format,
                    "column": column,
                })
            }
            SparkError::Arrow(e) => {
                serde_json::json!({
                    "message": e.to_string(),
                })
            }
            SparkError::Internal(msg) => {
                serde_json::json!({
                    "message": msg,
                })
            }
            // Simple errors with no parameters
            _ => serde_json::json!({}),
        }
    }

    /// Returns the appropriate Spark exception class for this error
    pub fn exception_class(&self) -> &'static str {
        match self {
            // ArithmeticException
            SparkError::DivideByZero
            | SparkError::RemainderByZero
            | SparkError::IntervalDividedByZero
            | SparkError::NumericValueOutOfRange { .. }
            | SparkError::NumericOutOfRange { .. } // Comet-specific extension
            | SparkError::ArithmeticOverflow { .. }
            | SparkError::IntegralDivideOverflow
            | SparkError::DecimalSumOverflow { .. }
            | SparkError::BinaryArithmeticOverflow { .. }
            | SparkError::IntervalArithmeticOverflowWithSuggestion { .. }
            | SparkError::IntervalArithmeticOverflowWithoutSuggestion
            | SparkError::DatetimeOverflow => "org/apache/spark/SparkArithmeticException",

            // CastOverflow gets special handling with CastOverflowException
            SparkError::CastOverFlow { .. } => "org/apache/spark/sql/comet/CastOverflowException",

            // NumberFormatException (for cast invalid input errors)
            SparkError::CastInvalidValue { .. } => "org/apache/spark/SparkNumberFormatException",

            // ArrayIndexOutOfBoundsException
            SparkError::InvalidArrayIndex { .. }
            | SparkError::InvalidElementAtIndex { .. }
            | SparkError::InvalidBitmapPosition { .. }
            | SparkError::InvalidIndexOfZero => "org/apache/spark/SparkArrayIndexOutOfBoundsException",

            // RuntimeException
            SparkError::MalformedVariant
            | SparkError::CannotParseDecimal
            | SparkError::DuplicatedMapKey { .. }
            | SparkError::NullMapKey
            | SparkError::MapKeyValueDiffSizes
            | SparkError::ExceedMapSizeLimit { .. }
            | SparkError::CollectionSizeLimitExceeded { .. }
            | SparkError::SequenceBatchTooLarge { .. } // Comet-specific extension
            | SparkError::NotNullAssertViolation { .. }
            | SparkError::ValueIsNull { .. } // Comet-specific extension
            | SparkError::UnexpectedPositiveValue { .. }
            | SparkError::UnexpectedNegativeValue { .. }
            | SparkError::InvalidRegexGroupIndex { .. }
            | SparkError::ScalarSubqueryTooManyRows
            | SparkError::MergeCardinalityViolation => "org/apache/spark/SparkRuntimeException",

            // DateTimeException
            SparkError::InvalidInputInCastToDatetime { .. }
            | SparkError::CannotParseTimestamp { .. }
            | SparkError::InvalidFractionOfSecond { .. }
            | SparkError::DatetimeFieldOutOfBounds { .. } => {
                "org/apache/spark/SparkDateTimeException"
            }

            // IllegalArgumentException
            SparkError::DatatypeCannotOrder { .. }
            | SparkError::InvalidUrl { .. }
            | SparkError::InvalidUtf8String { .. }
            | SparkError::IllegalDayOfWeek { .. }
            | SparkError::SequenceIllegalBoundaries { .. } => {
                "org/apache/spark/SparkIllegalArgumentException"
            }

            // FileNotFound - will be converted to SparkFileNotFoundException by the shim
            SparkError::FileNotFound { .. } => "org/apache/spark/SparkException",

            // DuplicateFieldCaseInsensitive - converted to SparkRuntimeException by the shim
            SparkError::DuplicateFieldCaseInsensitive { .. } => {
                "org/apache/spark/SparkRuntimeException"
            }

            // DuplicateFieldByFieldId - converted to SparkRuntimeException by the shim
            // (Spark's `foundDuplicateFieldInFieldIdLookupModeError` returns SparkRuntimeException)
            SparkError::DuplicateFieldByFieldId { .. } => "org/apache/spark/SparkRuntimeException",

            // ParquetMissingFieldIds - converted to the plain RuntimeException Spark's
            // ParquetReadSupport throws when the file lacks field ids and
            // `spark.sql.parquet.fieldId.read.ignoreMissing=false`. The 4.x shim wraps it in
            // the FAILED_READ_FILE SparkException Spark 4 raises at the task boundary.
            SparkError::ParquetMissingFieldIds { .. } => "java/lang/RuntimeException",

            // ParquetSchemaConvert - converted to SchemaColumnConvertNotSupportedException by the shim
            SparkError::ParquetSchemaConvert { .. } => {
                "org/apache/spark/sql/execution/datasources/SchemaColumnConvertNotSupportedException"
            }

            // CannotReadFile - converted to a FAILED_READ_FILE SparkException by the shim
            // (QueryExecutionErrors.cannotReadFilesError).
            SparkError::CannotReadFile { .. } => "org/apache/spark/SparkException",

            // ReadAncientDatetime - converted to SparkUpgradeException by the shim
            // (DataSourceUtils.newRebaseExceptionInRead).
            SparkError::ReadAncientDatetime { .. } => "org/apache/spark/SparkUpgradeException",

            // Generic errors
            SparkError::Arrow(_) | SparkError::Internal(_) => "org/apache/spark/SparkException",
        }
    }

    /// Returns the Spark error class code for this error
    pub(crate) fn error_class(&self) -> Option<&'static str> {
        match self {
            SparkError::MalformedVariant => Some("MALFORMED_VARIANT"),
            // Cast errors
            SparkError::CastInvalidValue { .. } => Some("CAST_INVALID_INPUT"),
            SparkError::InvalidInputInCastToDatetime { .. } => Some("CAST_INVALID_INPUT"),
            SparkError::CastOverFlow { .. } => Some("CAST_OVERFLOW"),
            SparkError::NumericValueOutOfRange { .. } => {
                Some("NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION")
            }
            SparkError::NumericOutOfRange { .. } => Some("NUMERIC_OUT_OF_SUPPORTED_RANGE"),
            SparkError::CannotParseDecimal => Some("CANNOT_PARSE_DECIMAL"),

            // Arithmetic errors
            SparkError::DivideByZero => Some("DIVIDE_BY_ZERO"),
            SparkError::RemainderByZero => Some("REMAINDER_BY_ZERO"),
            SparkError::IntervalDividedByZero => Some("INTERVAL_DIVIDED_BY_ZERO"),
            SparkError::ArithmeticOverflow { .. } => Some("ARITHMETIC_OVERFLOW"),
            SparkError::IntegralDivideOverflow => Some("ARITHMETIC_OVERFLOW"),
            SparkError::DecimalSumOverflow { .. } => Some("ARITHMETIC_OVERFLOW"),
            SparkError::BinaryArithmeticOverflow { .. } => Some("BINARY_ARITHMETIC_OVERFLOW"),
            SparkError::IntervalArithmeticOverflowWithSuggestion { .. } => {
                Some("INTERVAL_ARITHMETIC_OVERFLOW")
            }
            SparkError::IntervalArithmeticOverflowWithoutSuggestion => {
                Some("INTERVAL_ARITHMETIC_OVERFLOW")
            }
            SparkError::DatetimeOverflow => Some("DATETIME_OVERFLOW"),

            // Array index errors
            SparkError::InvalidArrayIndex { .. } => Some("INVALID_ARRAY_INDEX"),
            SparkError::InvalidElementAtIndex { .. } => Some("INVALID_ARRAY_INDEX_IN_ELEMENT_AT"),
            SparkError::InvalidBitmapPosition { .. } => Some("INVALID_BITMAP_POSITION"),
            SparkError::InvalidIndexOfZero => Some("INVALID_INDEX_OF_ZERO"),

            // Map/Collection errors
            SparkError::DuplicatedMapKey { .. } => Some("DUPLICATED_MAP_KEY"),
            SparkError::NullMapKey => Some("NULL_MAP_KEY"),
            SparkError::MapKeyValueDiffSizes => Some("MAP_KEY_VALUE_DIFF_SIZES"),
            SparkError::ExceedMapSizeLimit { .. } => Some("EXCEED_LIMIT_LENGTH"),
            SparkError::CollectionSizeLimitExceeded { .. } => {
                Some("COLLECTION_SIZE_LIMIT_EXCEEDED")
            }
            SparkError::SequenceIllegalBoundaries { .. } => Some("_LEGACY_ERROR_TEMP_3243"),

            // Comet-specific: no Spark error class, the shim builds the message itself.
            SparkError::SequenceBatchTooLarge { .. } => None,

            // Null validation errors
            SparkError::NotNullAssertViolation { .. } => Some("NOT_NULL_ASSERT_VIOLATION"),
            SparkError::ValueIsNull { .. } => Some("VALUE_IS_NULL"),

            // DateTime errors
            SparkError::CannotParseTimestamp { .. } => Some("CANNOT_PARSE_TIMESTAMP"),
            SparkError::InvalidFractionOfSecond { .. } => Some("INVALID_FRACTION_OF_SECOND"),
            SparkError::IllegalDayOfWeek { .. } => Some("ILLEGAL_DAY_OF_WEEK"),
            SparkError::DatetimeFieldOutOfBounds { .. } => Some("DATETIME_FIELD_OUT_OF_BOUNDS"),

            // String/UTF8 errors
            SparkError::InvalidUtf8String { .. } => Some("INVALID_UTF8_STRING"),

            // Function parameter errors
            SparkError::UnexpectedPositiveValue { .. } => Some("UNEXPECTED_POSITIVE_VALUE"),
            SparkError::UnexpectedNegativeValue { .. } => Some("UNEXPECTED_NEGATIVE_VALUE"),

            // Regex errors
            SparkError::InvalidRegexGroupIndex { .. } => Some("INVALID_PARAMETER_VALUE"),

            // URL errors
            SparkError::InvalidUrl { .. } => Some("INVALID_URL"),

            // Unsupported operation errors
            SparkError::DatatypeCannotOrder { .. } => Some("DATATYPE_CANNOT_ORDER"),

            // Subquery errors
            SparkError::ScalarSubqueryTooManyRows => Some("SCALAR_SUBQUERY_TOO_MANY_ROWS"),

            // MERGE INTO errors
            SparkError::MergeCardinalityViolation => Some("MERGE_CARDINALITY_VIOLATION"),

            // File not found
            SparkError::FileNotFound { .. } => Some("_LEGACY_ERROR_TEMP_2055"),

            // Duplicate field in case-insensitive mode
            SparkError::DuplicateFieldCaseInsensitive { .. } => Some("_LEGACY_ERROR_TEMP_2093"),

            // Duplicate field id in id-lookup mode
            SparkError::DuplicateFieldByFieldId { .. } => Some("_LEGACY_ERROR_TEMP_2094"),

            // ParquetMissingFieldIds is a plain RuntimeException with no error class. The 4.x
            // shim supplies FAILED_READ_FILE itself, so none is exposed here.
            SparkError::ParquetMissingFieldIds { .. } => None,

            // Parquet schema mismatch — translated to SchemaColumnConvertNotSupportedException
            // by the JVM shim. The shim wraps it in the version-appropriate
            // SparkException error class, so no error class is exposed here.
            SparkError::ParquetSchemaConvert { .. } => None,

            // CannotReadFile — the JVM shim wraps it via cannotReadFilesError, which supplies the
            // FAILED_READ_FILE error class, so none is exposed here.
            SparkError::CannotReadFile { .. } => None,

            // ReadAncientDatetime - set by DataSourceUtils.newRebaseExceptionInRead in the shim.
            SparkError::ReadAncientDatetime { .. } => {
                Some("INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME")
            }

            // Generic errors (no error class)
            SparkError::Arrow(_) | SparkError::Internal(_) => None,
        }
    }
}

pub type SparkResult<T> = Result<T, SparkError>;

/// Convert decimal overflow to SparkError::NumericValueOutOfRange.
pub fn decimal_overflow_error(value: i128, precision: u8, scale: i8) -> SparkError {
    SparkError::NumericValueOutOfRange {
        value: value.to_string(),
        precision,
        scale,
    }
}

/// Wrapper that adds QueryContext to SparkError
///
/// This allows attaching SQL context information (query text, line/position, object name) to errors
#[derive(Debug, Clone)]
pub struct SparkErrorWithContext {
    /// The underlying SparkError
    pub error: SparkError,
    /// Optional QueryContext for SQL location information
    pub context: Option<Arc<crate::QueryContext>>,
}

impl SparkErrorWithContext {
    /// Create a SparkErrorWithContext without context
    pub fn new(error: SparkError) -> Self {
        Self {
            error,
            context: None,
        }
    }

    /// Create a SparkErrorWithContext with QueryContext
    pub fn with_context(error: SparkError, context: Arc<crate::QueryContext>) -> Self {
        Self {
            error,
            context: Some(context),
        }
    }

    /// Serialize to JSON including optional context field
    pub fn to_json(&self) -> String {
        let mut json_obj = serde_json::json!({
            "errorType": self.error.error_type_name(),
            "errorClass": self.error.error_class().unwrap_or(""),
            "params": self.error.params_as_json(),
        });

        if let Some(ctx) = &self.context {
            // Serialize context fields
            json_obj["context"] = serde_json::json!({
                "sqlText": ctx.sql_text.as_str(),
                "startIndex": ctx.start_index,
                "stopIndex": ctx.stop_index,
                "objectType": ctx.object_type,
                "objectName": ctx.object_name,
                "line": ctx.line,
                "startPosition": ctx.start_position,
            });

            // Add formatted summary
            json_obj["summary"] = serde_json::json!(ctx.format_summary());
        }

        serde_json::to_string(&json_obj).unwrap_or_else(|e| {
            format!(
                "{{\"errorType\":\"SerializationError\",\"message\":\"{}\"}}",
                e
            )
        })
    }
}

impl std::fmt::Display for SparkErrorWithContext {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.error)?;
        if let Some(ctx) = &self.context {
            write!(f, "\n{}", ctx.format_summary())?;
        }
        Ok(())
    }
}

impl std::error::Error for SparkErrorWithContext {}

impl From<SparkError> for SparkErrorWithContext {
    fn from(error: SparkError) -> Self {
        SparkErrorWithContext::new(error)
    }
}

impl From<SparkErrorWithContext> for DataFusionError {
    fn from(value: SparkErrorWithContext) -> Self {
        DataFusionError::External(Box::new(value))
    }
}

impl From<ArrowError> for SparkError {
    fn from(value: ArrowError) -> Self {
        SparkError::Arrow(Arc::new(value))
    }
}

impl From<SparkError> for DataFusionError {
    fn from(value: SparkError) -> Self {
        DataFusionError::External(Box::new(value))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_divide_by_zero_json() {
        let error = SparkError::DivideByZero;
        let json = error.to_json();

        assert!(json.contains("\"errorType\":\"DivideByZero\""));
        assert!(json.contains("\"errorClass\":\"DIVIDE_BY_ZERO\""));

        // Verify it's valid JSON
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "DivideByZero");
        assert_eq!(parsed["errorClass"], "DIVIDE_BY_ZERO");
    }

    #[test]
    fn test_remainder_by_zero_json() {
        let error = SparkError::RemainderByZero;
        let json = error.to_json();

        assert!(json.contains("\"errorType\":\"RemainderByZero\""));
        assert!(json.contains("\"errorClass\":\"REMAINDER_BY_ZERO\""));
    }

    #[test]
    fn test_arithmetic_overflow_suggestion_json_and_display() {
        for function in ["", "try_add", "try_subtract", "try_multiply"] {
            let error = SparkError::ArithmeticOverflow {
                from_type: "long".to_string(),
                function_name: function.to_string(),
            };
            let parsed: serde_json::Value = serde_json::from_str(&error.to_json()).unwrap();
            assert_eq!(parsed["errorClass"], "ARITHMETIC_OVERFLOW");
            assert_eq!(parsed["params"]["fromType"], "long");
            assert_eq!(parsed["params"]["functionName"], function);
            let hint = if function.is_empty() {
                String::new()
            } else {
                format!(" Use '{function}' to tolerate overflow and return NULL instead.")
            };
            assert_eq!(error.to_string(), format!(
                "[ARITHMETIC_OVERFLOW] long overflow.{hint} If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error."
            ));
        }
    }

    #[test]
    fn test_binary_overflow_json() {
        let error = SparkError::BinaryArithmeticOverflow {
            value1: "32767".to_string(),
            symbol: "+".to_string(),
            value2: "1".to_string(),
            function_name: "try_add".to_string(),
        };
        let json = error.to_json();

        // Verify it's valid JSON
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "BinaryArithmeticOverflow");
        assert_eq!(parsed["errorClass"], "BINARY_ARITHMETIC_OVERFLOW");
        assert_eq!(parsed["params"]["value1"], "32767");
        assert_eq!(parsed["params"]["symbol"], "+");
        assert_eq!(parsed["params"]["value2"], "1");
        assert_eq!(parsed["params"]["functionName"], "try_add");
    }

    #[test]
    fn test_invalid_array_index_json() {
        let error = SparkError::InvalidArrayIndex {
            index_value: 10,
            array_size: 3,
        };
        let json = error.to_json();

        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "InvalidArrayIndex");
        assert_eq!(parsed["errorClass"], "INVALID_ARRAY_INDEX");
        assert_eq!(parsed["params"]["indexValue"], 10);
        assert_eq!(parsed["params"]["arraySize"], 3);
    }

    #[test]
    fn test_numeric_value_out_of_range_json() {
        let error = SparkError::NumericValueOutOfRange {
            value: "999.99".to_string(),
            precision: 5,
            scale: 2,
        };
        let json = error.to_json();

        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "NumericValueOutOfRange");
        assert_eq!(
            parsed["errorClass"],
            "NUMERIC_VALUE_OUT_OF_RANGE.WITH_SUGGESTION"
        );
        assert_eq!(parsed["params"]["value"], "999.99");
        assert_eq!(parsed["params"]["precision"], 5);
        assert_eq!(parsed["params"]["scale"], 2);
    }

    #[test]
    fn test_cast_invalid_value_json() {
        let error = SparkError::CastInvalidValue {
            value: "abc".to_string(),
            from_type: "STRING".to_string(),
            to_type: "INT".to_string(),
        };
        let json = error.to_json();

        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "CastInvalidValue");
        assert_eq!(parsed["errorClass"], "CAST_INVALID_INPUT");
        assert_eq!(parsed["params"]["value"], "abc");
        assert_eq!(parsed["params"]["fromType"], "STRING");
        assert_eq!(parsed["params"]["toType"], "INT");
    }

    #[test]
    fn test_duplicated_map_key_json() {
        let error = SparkError::DuplicatedMapKey {
            key: "duplicate_key".to_string(),
        };
        let json = error.to_json();

        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "DuplicatedMapKey");
        assert_eq!(parsed["errorClass"], "DUPLICATED_MAP_KEY");
        assert_eq!(parsed["params"]["key"], "duplicate_key");
    }

    #[test]
    fn test_null_map_key_json() {
        let error = SparkError::NullMapKey;
        let json = error.to_json();

        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["errorType"], "NullMapKey");
        assert_eq!(parsed["errorClass"], "NULL_MAP_KEY");
        // Params should be an empty object
        assert_eq!(parsed["params"], serde_json::json!({}));
    }

    #[test]
    fn test_error_class_mapping() {
        // Test that error_class() returns the correct error class
        assert_eq!(
            SparkError::DivideByZero.error_class(),
            Some("DIVIDE_BY_ZERO")
        );
        assert_eq!(
            SparkError::RemainderByZero.error_class(),
            Some("REMAINDER_BY_ZERO")
        );
        assert_eq!(
            SparkError::InvalidArrayIndex {
                index_value: 0,
                array_size: 0
            }
            .error_class(),
            Some("INVALID_ARRAY_INDEX")
        );
        assert_eq!(SparkError::NullMapKey.error_class(), Some("NULL_MAP_KEY"));
        assert_eq!(
            SparkError::read_ancient_datetime("d", false).error_class(),
            Some("INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME")
        );
    }

    #[test]
    fn test_exception_class_mapping() {
        // Test that exception_class() returns the correct Java exception class
        assert_eq!(
            SparkError::DivideByZero.exception_class(),
            "org/apache/spark/SparkArithmeticException"
        );
        assert_eq!(
            SparkError::InvalidArrayIndex {
                index_value: 0,
                array_size: 0
            }
            .exception_class(),
            "org/apache/spark/SparkArrayIndexOutOfBoundsException"
        );
        assert_eq!(
            SparkError::NullMapKey.exception_class(),
            "org/apache/spark/SparkRuntimeException"
        );
        assert_eq!(
            SparkError::read_ancient_datetime("d", false).exception_class(),
            "org/apache/spark/SparkUpgradeException"
        );
    }

    #[test]
    fn test_read_ancient_datetime_json() {
        for (is_int96, format) in [(false, "Parquet"), (true, "Parquet INT96")] {
            let error = SparkError::read_ancient_datetime("ts", is_int96);
            assert_eq!(error.error_type_name(), "ReadAncientDatetime");

            let parsed: serde_json::Value = serde_json::from_str(&error.to_json()).unwrap();
            assert_eq!(parsed["errorType"], "ReadAncientDatetime");
            assert_eq!(
                parsed["errorClass"],
                "INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME"
            );
            assert_eq!(
                parsed["params"],
                serde_json::json!({"format": format, "column": "ts"})
            );
        }
    }
}