errors.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.

//! Common Parquet errors and macros.

use arrow::error::ArrowError;
use datafusion::common::DataFusionError;
use datafusion_comet_common::{SparkError, SparkErrorWithContext};
use jni::errors::{Exception, ToException};
use regex::Regex;

use std::{
    any::Any,
    convert,
    fmt::Write,
    panic::UnwindSafe,
    result, str,
    str::Utf8Error,
    sync::{Arc, Mutex},
};

// This is just a pointer. We'll be returning it from our function. We
// can't return one of the objects with lifetime information because the
// lifetime checker won't let us.
use jni::sys::{jboolean, jbyte, jchar, jdouble, jfloat, jint, jlong, jobject, jshort};

use jni::objects::{Global, JThrowable};
use jni::{strings::JNIString, Env, EnvUnowned, Outcome};
use lazy_static::lazy_static;
use parquet::errors::ParquetError;
use thiserror::Error;

lazy_static! {
    static ref PANIC_BACKTRACE: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
}

/// Error returned during executing operators.
#[derive(thiserror::Error, Debug)]
pub enum ExecutionError {
    /// Simple error
    #[allow(dead_code)]
    #[error("General execution error with reason: {0}.")]
    GeneralError(String),

    /// Error when deserializing an operator.
    #[error("Fail to deserialize to native operator with reason: {0}.")]
    DeserializeError(String),

    /// Error when processing Arrow array.
    #[error("Fail to process Arrow array with reason: {0}.")]
    ArrowError(String),

    /// DataFusion error
    #[error("Error from DataFusion: {0}.")]
    DataFusionError(String),

    #[error("{class}: {msg}")]
    JavaException {
        class: String,
        msg: String,
        throwable: Global<JThrowable<'static>>,
    },
}

#[derive(thiserror::Error, Debug)]
pub enum CometError {
    #[error("Configuration Error: {0}")]
    Config(String),

    #[error("{0}")]
    NullPointer(String),

    #[error("Out of bounds{0}")]
    IndexOutOfBounds(usize),

    #[error("Comet Internal Error: {0}")]
    Internal(String),

    #[error(transparent)]
    Arrow {
        #[from]
        source: ArrowError,
    },

    #[error(transparent)]
    Parquet {
        #[from]
        source: ParquetError,
    },

    #[error(transparent)]
    Expression {
        #[from]
        source: ExpressionError,
    },

    #[error(transparent)]
    Execution {
        #[from]
        source: ExecutionError,
    },

    #[error(transparent)]
    IO {
        #[from]
        source: std::io::Error,
    },

    #[error(transparent)]
    NumberIntFormat {
        #[from]
        source: std::num::ParseIntError,
    },

    #[error(transparent)]
    NumberFloatFormat {
        #[from]
        source: std::num::ParseFloatError,
    },
    #[error(transparent)]
    BoolFormat {
        #[from]
        source: std::str::ParseBoolError,
    },
    #[error(transparent)]
    Format {
        #[from]
        source: Utf8Error,
    },

    #[error(transparent)]
    JNI {
        #[from]
        source: jni::errors::Error,
    },

    #[error("{msg}")]
    Panic { msg: String },

    #[error("{msg}")]
    DataFusion {
        msg: String,
        #[source]
        source: DataFusionError,
    },

    /// Wraps a SparkError directly, allowing Comet to throw Spark-compatible exceptions
    /// that Spark would return
    #[error(transparent)]
    Spark(SparkError),

    #[error("{class}: {msg}")]
    JavaException {
        class: String,
        msg: String,
        throwable: Global<JThrowable<'static>>,
    },
}

pub fn init() {
    std::panic::set_hook(Box::new(|panic_info| {
        // Log the panic message and location to stderr so it is visible in CI logs
        // even if the exception message is lost crossing the FFI boundary
        eprintln!("Comet native panic: {panic_info}");
        // Capture the backtrace for a panic
        *PANIC_BACKTRACE.lock().unwrap() =
            Some(std::backtrace::Backtrace::force_capture().to_string());
    }));
}

/// Converts the results from `panic::catch_unwind` (e.g. a panic) to a `CometError`
impl convert::From<Box<dyn Any + Send>> for CometError {
    fn from(e: Box<dyn Any + Send>) -> Self {
        CometError::Panic {
            msg: match e.downcast_ref::<&str>() {
                Some(s) => s.to_string(),
                None => match e.downcast_ref::<String>() {
                    Some(msg) => msg.to_string(),
                    None => "unknown panic".to_string(),
                },
            },
        }
    }
}

impl From<DataFusionError> for CometError {
    fn from(value: DataFusionError) -> Self {
        CometError::DataFusion {
            msg: value.message().to_string(),
            source: value,
        }
    }
}

impl From<CometError> for DataFusionError {
    fn from(value: CometError) -> Self {
        match value {
            CometError::DataFusion { msg: _, source } => source,
            // Preserve the original Java throwable (e.g. a SparkRuntimeException raised by Spark's
            // own codegen inside the JVM UDF kernel) as an `External` error so it survives the trip
            // back through DataFusion and can be re-thrown with its exact type at the JNI boundary.
            // Flattening it to a string here would surface it as a generic CometNativeException.
            value @ CometError::JavaException { .. } => DataFusionError::External(Box::new(value)),
            _ => DataFusionError::Execution(value.to_string()),
        }
    }
}

impl From<CometError> for ParquetError {
    fn from(value: CometError) -> Self {
        match value {
            CometError::Parquet { source } => source,
            _ => ParquetError::General(value.to_string()),
        }
    }
}

impl From<CometError> for ExecutionError {
    fn from(value: CometError) -> Self {
        match value {
            CometError::Execution { source } => source,
            CometError::JavaException {
                class,
                msg,
                throwable,
            } => ExecutionError::JavaException {
                class,
                msg,
                throwable,
            },
            _ => ExecutionError::GeneralError(value.to_string()),
        }
    }
}

impl From<prost::DecodeError> for ExpressionError {
    fn from(error: prost::DecodeError) -> ExpressionError {
        ExpressionError::Deserialize(error.to_string())
    }
}

impl From<prost::UnknownEnumValue> for ExpressionError {
    fn from(error: prost::UnknownEnumValue) -> ExpressionError {
        ExpressionError::Deserialize(error.to_string())
    }
}

impl From<prost::DecodeError> for ExecutionError {
    fn from(error: prost::DecodeError) -> ExecutionError {
        ExecutionError::DeserializeError(error.to_string())
    }
}

impl From<prost::UnknownEnumValue> for ExecutionError {
    fn from(error: prost::UnknownEnumValue) -> ExecutionError {
        ExecutionError::DeserializeError(error.to_string())
    }
}

impl From<ArrowError> for ExecutionError {
    fn from(error: ArrowError) -> ExecutionError {
        ExecutionError::ArrowError(error.to_string())
    }
}

impl From<ArrowError> for ExpressionError {
    fn from(error: ArrowError) -> ExpressionError {
        ExpressionError::ArrowError(error.to_string())
    }
}

impl From<ExpressionError> for ArrowError {
    fn from(error: ExpressionError) -> ArrowError {
        ArrowError::ComputeError(error.to_string())
    }
}

impl From<DataFusionError> for ExecutionError {
    fn from(value: DataFusionError) -> Self {
        ExecutionError::DataFusionError(value.message().to_string())
    }
}

impl From<ExecutionError> for DataFusionError {
    fn from(value: ExecutionError) -> Self {
        DataFusionError::Execution(value.to_string())
    }
}

impl From<ExpressionError> for DataFusionError {
    fn from(value: ExpressionError) -> Self {
        DataFusionError::Execution(value.to_string())
    }
}

impl jni::errors::ToException for CometError {
    fn to_exception(&self) -> Exception {
        match self {
            CometError::IndexOutOfBounds(..) => Exception {
                class: "java/lang/IndexOutOfBoundsException".to_string(),
                msg: self.to_string(),
            },
            CometError::NullPointer(..) => Exception {
                class: "java/lang/NullPointerException".to_string(),
                msg: self.to_string(),
            },
            CometError::NumberIntFormat { source: s } => Exception {
                class: "java/lang/NumberFormatException".to_string(),
                msg: s.to_string(),
            },
            CometError::NumberFloatFormat { source: s } => Exception {
                class: "java/lang/NumberFormatException".to_string(),
                msg: s.to_string(),
            },
            CometError::IO { .. } => Exception {
                class: "java/io/IOException".to_string(),
                msg: self.to_string(),
            },
            CometError::Parquet { .. } => Exception {
                class: "org/apache/comet/ParquetRuntimeException".to_string(),
                msg: self.to_string(),
            },
            CometError::Spark(spark_err) => Exception {
                class: spark_err.exception_class().to_string(),
                msg: spark_err.to_string(),
            },
            _other => Exception {
                class: "org/apache/comet/CometNativeException".to_string(),
                msg: self.to_string(),
            },
        }
    }
}

/// Error returned when there is an error during executing an expression.
#[derive(thiserror::Error, Debug)]
pub enum ExpressionError {
    /// Simple error
    #[error("General expression error with reason {0}.")]
    General(String),

    /// Deserialization error
    #[error("Fail to deserialize to native expression with reason {0}.")]
    Deserialize(String),

    /// Evaluation error
    #[error("Fail to evaluate native expression with reason {0}.")]
    Evaluation(String),

    /// Error when processing Arrow array.
    #[error("Fail to process Arrow array with reason {0}.")]
    ArrowError(String),
}

/// A specialized `Result` for Comet errors.
pub type CometResult<T> = result::Result<T, CometError>;

// ----------------------------------------------------------------------
// Convenient macros for different errors

#[macro_export]
macro_rules! general_err {
    ($fmt:expr, $($args:expr),*) => ($crate::errors::CometError::from(parquet::errors::ParquetError::General(format!($fmt, $($args),*))));
}

/// Returns the "default value" for a type.  This is used for JNI code in order to facilitate
/// returning a value in cases where an exception is thrown.  This value will never be used, as the
/// JVM will note the pending exception.
///
/// Default values are often some kind of initial value, identity value, or anything else that
/// may make sense as a default.
///
/// NOTE: We can't just use [Default] since both the trait and the object are defined in other
/// crates.
/// See [Rust Compiler Error Index - E0117](https://doc.rust-lang.org/error-index.html#E0117)
pub trait JNIDefault {
    fn default() -> Self;
}

impl JNIDefault for jboolean {
    fn default() -> jboolean {
        false
    }
}

impl JNIDefault for jbyte {
    fn default() -> jbyte {
        0
    }
}

impl JNIDefault for jchar {
    fn default() -> jchar {
        0
    }
}

impl JNIDefault for jdouble {
    fn default() -> jdouble {
        0.0
    }
}

impl JNIDefault for jfloat {
    fn default() -> jfloat {
        0.0
    }
}

impl JNIDefault for jint {
    fn default() -> jint {
        0
    }
}

impl JNIDefault for jlong {
    fn default() -> jlong {
        0
    }
}

/// The "default value" for all returned objects, such as [jstring], [jlongArray], etc.
impl JNIDefault for jobject {
    fn default() -> jobject {
        std::ptr::null_mut()
    }
}

impl JNIDefault for jshort {
    fn default() -> jshort {
        0
    }
}

impl JNIDefault for () {
    fn default() {}
}

// Unwrap the result returned from `panic::catch_unwind` when `Ok`, otherwise throw a
// `RuntimeException` back to the calling Java.  Since a return result is required, use `JNIDefault`
// to create a reasonable result.  This returned default value will be ignored due to the exception.
pub fn unwrap_or_throw_default<T: JNIDefault>(
    env: &mut Env,
    result: std::result::Result<T, CometError>,
) -> T {
    match result {
        Ok(value) => value,
        Err(err) => {
            let backtrace = match err {
                CometError::Panic { msg: _ } => PANIC_BACKTRACE.lock().unwrap().take(),
                _ => None,
            };
            throw_exception(env, &err, backtrace);
            T::default()
        }
    }
}

/// Payload recovered from a DataFusionError chain.
enum SparkPayload<'a> {
    JavaException(&'a Global<JThrowable<'static>>),
    WithContext(&'a SparkErrorWithContext),
    Bare(&'a SparkError),
}

/// Recursively unwrap `DataFusionError::Context` and `DataFusionError::External` layers
/// until a Spark-typed payload is found, or return `None`.
///
/// DataFusion 53+ wraps errors with `.context(...)` which produces
/// `DataFusionError::Context(description, Box<inner>)`. The JNI bridge must look
/// through this extra layer — and through any doubly-nested `External` — to reach
/// the `SparkError` / `SparkErrorWithContext` that carries the structured exception.
fn extract_spark_payload(err: &DataFusionError) -> Option<SparkPayload<'_>> {
    match err {
        DataFusionError::External(e) => {
            if let Some(CometError::JavaException { throwable, .. }) =
                e.downcast_ref::<CometError>()
            {
                return Some(SparkPayload::JavaException(throwable));
            }
            if let Some(ctx) = e.downcast_ref::<SparkErrorWithContext>() {
                return Some(SparkPayload::WithContext(ctx));
            }
            if let Some(spark) = e.downcast_ref::<SparkError>() {
                return Some(SparkPayload::Bare(spark));
            }
            // Recurse: External may wrap another DataFusionError (double-wrapping).
            if let Some(inner_df) = e.downcast_ref::<DataFusionError>() {
                return extract_spark_payload(inner_df);
            }
            None
        }
        // DataFusion 53 adds context via `.context(description)` which wraps the error in
        // Context(description, Box<inner>). Strip the context wrapper and recurse.
        DataFusionError::Context(_, inner) => extract_spark_payload(inner),
        _ => None,
    }
}

fn throw_exception(env: &mut Env, error: &CometError, backtrace: Option<String>) {
    // If there isn't already an exception?
    if !env.exception_check() {
        // ... then throw new exception
        // Note: in jni 0.22.x, throw/throw_new return Err(JavaException) on success
        // (to signal the pending exception to Rust callers via `?`). We discard the
        // result here because we're in an error-handling path and just need the
        // exception to be pending in the JVM.
        let _ = match error {
            CometError::JavaException {
                class: _,
                msg: _,
                throwable,
            } => env.throw(throwable),
            CometError::Execution {
                source:
                    ExecutionError::JavaException {
                        class: _,
                        msg: _,
                        throwable,
                    },
            } => env.throw(throwable),
            // Handle all DataFusion errors, including Context-wrapped chains.
            // `extract_spark_payload` recurses through Context / nested External layers to
            // find the Spark-typed payload, so this arm covers:
            //   - DataFusionError::External(SparkErrorWithContext)      (normal path)
            //   - DataFusionError::External(SparkError)                 (no query context)
            //   - DataFusionError::Context(_, External(SparkError))     (DF53 context wrapping)
            //   - DataFusionError::External(External(SparkError))       (double wrapping)
            CometError::DataFusion { msg: _, source } => {
                match extract_spark_payload(source) {
                    Some(SparkPayload::JavaException(throwable)) => {
                        // A Java exception captured inside a JVM UDF kernel (e.g. Spark codegen
                        // raising INVALID_REGEXP_REPLACE). Re-throw the original throwable so
                        // callers see the exact Spark exception type.
                        env.throw(throwable)
                    }
                    Some(SparkPayload::WithContext(ctx)) => {
                        let json_message = ctx.to_json();
                        env.throw_new(
                            jni::jni_str!(
                                "org/apache/comet/exceptions/CometQueryExecutionException"
                            ),
                            JNIString::new(json_message),
                        )
                    }
                    Some(SparkPayload::Bare(spark_error)) => {
                        throw_spark_error_as_json(env, spark_error)
                    }
                    None => {
                        if let Some(spark_error) = try_classify_file_read_error(source) {
                            throw_spark_error_as_json(env, &spark_error)
                        } else {
                            throw_generic_exception(env, error, backtrace)
                        }
                    }
                }
            }
            // Handle direct SparkError - serialize to JSON
            CometError::Spark(spark_error) => throw_spark_error_as_json(env, spark_error),
            _ => throw_generic_exception(env, error, backtrace),
        };
    }
}

/// Generic fallback throw for an error that isn't a structured `SparkError`. Recognises a
/// file-not-found arriving through non-typed wrapping paths and duplicate-field errors; otherwise
/// throws the error's natural JVM exception (with the captured backtrace when available).
fn throw_generic_exception(
    env: &mut Env,
    error: &CometError,
    backtrace: Option<String>,
) -> jni::errors::Result<()> {
    let error_msg = error.to_string();
    // A file-not-found that arrived through a non-typed wrapping path (the typed classification
    // is handled by `try_classify_file_read_error`).
    if error_msg.contains("not found") && error_msg.contains("No such file or directory") {
        let spark_error = SparkError::FileNotFound { message: error_msg };
        throw_spark_error_as_json(env, &spark_error)
    } else if let Some(spark_error) = try_convert_duplicate_field_error(&error_msg) {
        throw_spark_error_as_json(env, &spark_error)
    } else {
        let exception = error.to_exception();
        match backtrace {
            Some(backtrace_string) => env.throw_new(
                JNIString::new(exception.class),
                JNIString::new(to_stacktrace_string(exception.msg, backtrace_string).unwrap()),
            ),
            _ => env.throw_new(
                JNIString::new(exception.class),
                JNIString::new(exception.msg),
            ),
        }
    }
}

/// Throws a CometQueryExecutionException with JSON-encoded SparkError
fn throw_spark_error_as_json(env: &mut Env, spark_error: &SparkError) -> jni::errors::Result<()> {
    // Serialize error to JSON
    let json_message = spark_error.to_json();

    // Throw CometQueryExecutionException with JSON message
    env.throw_new(
        jni::jni_str!("org/apache/comet/exceptions/CometQueryExecutionException"),
        JNIString::new(json_message),
    )
}

/// Classify a `DataFusionError` as a per-file read failure by TYPED variant (not message text),
/// returning `SparkError::CannotReadFile` if so. This is the structured replacement for the
/// previous JVM-side substring matching on error prose.
///
/// A file-read failure is any of:
///   - `ParquetError` (corrupt footer/page, EOF, "failed to fill whole buffer", etc.)
///   - `ObjectStore` (truncated/empty/deleted file, range errors) -- `NotFound` carries the path
///   - `ArrowError`, when it wraps a `ParquetError` (the parquet reader surfaces some failures as
///     `ArrowError::ParquetError`)
///   - `IoError` (filesystem read failures)
///
/// `Context`/`Shared` wrappers are unwrapped recursively. Note we do NOT match `Execution`/
/// `Internal`/`External`-string or `object_store::Error::Generic`: those also carry non-file
/// errors (e.g. "Hdfs support is not enabled in this build") that must surface as-is.
///
/// `file_path` is populated from `object_store::Error::NotFound { path, .. }` when available;
/// otherwise it is left empty and the JVM side fills it from the per-task file list.
fn try_classify_file_read_error(error: &DataFusionError) -> Option<SparkError> {
    use datafusion::common::DataFusionError as DFE;
    match error {
        // A pushed-down filter predicate that throws while the scan is reading (e.g. an ANSI
        // divide-by-zero in a `WHERE`, now reachable by default since Scala UDF codegen dispatch
        // landed) is an EXPRESSION failure, not a file-read failure, and must not be relabelled
        // FAILED_READ_FILE. DataFusion's row filter returns such a failure as `ArrowError`
        // (`ComputeError`), which the parquet reader then wraps as `ParquetError::External(<arrow>)`.
        // Genuine corrupt/truncated/missing-file errors are `ParquetError::General`/`EOF`/
        // `External(io|object_store)` and never wrap an `ArrowError`, so this TYPED check (not the
        // error-message text, which DataFusion produces via `{:?}`) tells the two apart. Bail so the
        // underlying error surfaces through the normal native-exception path.
        DFE::ParquetError(pe) if parquet_external_wraps_arrow_error(pe) => None,
        // A genuinely-missing file (object_store NotFound) is distinct from a corrupt/truncated
        // one: Spark surfaces it as `readCurrentFileNotFoundError` ("It is possible the underlying
        // files have been updated."), not `cannotReadFilesError`. The NotFound may arrive directly
        // (`DFE::ObjectStore`) or wrapped by the parquet reader as `ParquetError::External(..)`, so
        // inspect the source chain. Delta's CDC-after-VACUUM read depends on this distinction. The
        // message fallback covers the cached reader, which flattens NotFound into a sourceless string.
        DFE::ParquetError(pe)
            if source_chain_has_object_store_not_found(pe.as_ref())
                || parquet_message_has_object_store_not_found(pe) =>
        {
            Some(SparkError::FileNotFound {
                message: error.to_string(),
            })
        }
        // NB: only ParquetError / ObjectStore / ArrowError(ParquetError) are treated as file reads.
        // A bare `IoError` is intentionally NOT classified here: a scan surfaces read failures as a
        // typed ParquetError or ObjectStore error, whereas an `IoError` can also originate from
        // non-scan paths (spill, shuffle), which must not be relabelled FAILED_READ_FILE.
        DFE::ParquetError(_) => Some(SparkError::CannotReadFile {
            file_path: String::new(),
            message: cannot_read_file_message(error),
        }),
        DFE::ObjectStore(e) => match e.as_ref() {
            datafusion::object_store::Error::NotFound { .. } => Some(SparkError::FileNotFound {
                message: error.to_string(),
            }),
            _ => Some(SparkError::CannotReadFile {
                file_path: String::new(),
                message: cannot_read_file_message(error),
            }),
        },
        // The parquet reader sometimes surfaces a failure as ArrowError::ParquetError.
        DFE::ArrowError(e, _) => match e.as_ref() {
            ArrowError::ParquetError(_) if source_chain_has_object_store_not_found(e.as_ref()) => {
                Some(SparkError::FileNotFound {
                    message: error.to_string(),
                })
            }
            ArrowError::ParquetError(_) => Some(SparkError::CannotReadFile {
                file_path: String::new(),
                message: cannot_read_file_message(error),
            }),
            _ => None,
        },
        // Unwrap context/shared wrappers and re-classify the inner error.
        DFE::Context(_, inner) => try_classify_file_read_error(inner),
        DFE::Shared(inner) => try_classify_file_read_error(inner),
        _ => None,
    }
}

/// True if `pe` is a `ParquetError::External` wrapping an `ArrowError`. DataFusion's parquet row
/// filter returns a pushed-down predicate's evaluation failure as an `ArrowError` (e.g.
/// `ComputeError` for an ANSI divide-by-zero), which the parquet reader then surfaces as
/// `ParquetError::External(<arrow error>)`. That is an expression failure that merely happened
/// during the scan, not a corrupt/truncated/missing file -- genuine read failures are
/// `ParquetError::General`/`EOF`/`External(io|object_store)` and never wrap an `ArrowError`. Matching
/// on the wrapped type (rather than the message text DataFusion builds with `{:?}`) keeps the
/// distinction robust to upstream message changes.
fn parquet_external_wraps_arrow_error(pe: &ParquetError) -> bool {
    matches!(pe, ParquetError::External(inner) if inner.downcast_ref::<ArrowError>().is_some())
}

/// True if `err` or any error in its `source()` chain is an `object_store` `NotFound` -- i.e. a
/// genuinely-missing file. Used to tell a missing file apart from a corrupt/truncated one: the
/// parquet reader wraps the object_store error as `ParquetError::External(..)`, so the typed
/// `NotFound` is only reachable by walking the source chain (we match the typed variant, never the
/// message text).
fn source_chain_has_object_store_not_found(err: &(dyn std::error::Error + 'static)) -> bool {
    let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err);
    while let Some(e) = current {
        if let Some(os) = e.downcast_ref::<datafusion::object_store::Error>() {
            if matches!(os, datafusion::object_store::Error::NotFound { .. }) {
                return true;
            }
        }
        current = e.source();
    }
    false
}

/// Fallback for the one path that loses the typed error: DataFusion's `CachedParquetFileReader`
/// flattens a missing-file NotFound into `ParquetError::General(format!(...))` with no `source()` to
/// walk. Match `object_store`'s NotFound `Display`, which is identical across stores (so it holds on
/// S3), unlike the OS-specific "No such file or directory".
fn parquet_message_has_object_store_not_found(pe: &ParquetError) -> bool {
    let msg = pe.to_string();
    msg.contains("Object at location") && msg.contains("not found")
}

/// Build the message for a `CannotReadFile` error. parquet-rs reports a bad magic / unreadable
/// footer as `"Invalid Parquet file. Corrupt footer"`, whereas Spark's own reader (and Spark's
/// `ParquetQuerySuite`) phrase it as `"<file> is not a Parquet file"`. Append Spark's phrasing so
/// the cause carries it; the outer `cannotReadFilesError` wrapper ("Encountered error while reading
/// file …") is unchanged, so this composes with Spark's tests without changing the FAILED_READ_FILE
/// wrapping. Other read failures (corrupt pages, EOF, IO) keep their native message verbatim.
fn cannot_read_file_message(error: &DataFusionError) -> String {
    let msg = error.to_string();
    if msg.contains("Invalid Parquet file") && !msg.contains("is not a Parquet file") {
        format!("{msg} (file is not a Parquet file)")
    } else {
        msg
    }
}

/// Try to convert a DataFusion "Unable to get field named" error into a SparkError.
/// DataFusion produces this error when reading Parquet files with duplicate field names
/// in case-insensitive mode. For example, if a Parquet file has columns "B" and "b",
/// DataFusion may deduplicate them and report: Unable to get field named "b". Valid
/// fields: ["A", "B"]. When the requested field has a case-insensitive match among the
/// valid fields, we convert this to Spark's _LEGACY_ERROR_TEMP_2093 error.
fn try_convert_duplicate_field_error(error_msg: &str) -> Option<SparkError> {
    // Match: Schema error: Unable to get field named "X". Valid fields: [...]
    lazy_static! {
        static ref FIELD_RE: Regex =
            Regex::new(r#"Unable to get field named "([^"]+)"\. Valid fields: \[(.+)\]"#).unwrap();
    }
    if let Some(caps) = FIELD_RE.captures(error_msg) {
        let requested_field = caps.get(1)?.as_str();
        let requested_lower = requested_field.to_lowercase();
        // Parse field names from the Valid fields list: ["A", "B"] or [A, B, b]
        let valid_fields_raw = caps.get(2)?.as_str();
        let all_fields: Vec<String> = valid_fields_raw
            .split(',')
            .map(|s| s.trim().trim_matches('"').to_string())
            .collect();
        // Find fields that match case-insensitively
        let mut matched: Vec<String> = all_fields
            .into_iter()
            .filter(|f| f.to_lowercase() == requested_lower)
            .collect();
        // Need at least one case-insensitive match to treat this as a duplicate field error.
        // DataFusion may deduplicate columns case-insensitively, so the valid fields list
        // might contain only one variant (e.g. "B" when file has both "B" and "b").
        // If requested field differs from the match, both existed in the original file.
        if matched.is_empty() {
            return None;
        }
        // Add the requested field name if it's not already in the list (different case)
        if !matched.iter().any(|f| f == requested_field) {
            matched.push(requested_field.to_string());
        }
        let required_field_name = requested_field.to_string();
        let matched_fields = format!("[{}]", matched.join(", "));
        Some(SparkError::DuplicateFieldCaseInsensitive {
            required_field_name,
            matched_fields,
        })
    } else {
        None
    }
}

#[derive(Debug, Error)]
enum StacktraceError {
    #[error("Unable to initialize message: {0}")]
    Message(String),
    #[error("Unable to initialize backtrace regex: {0}")]
    Regex(#[from] regex::Error),
    #[error("Required field missing: {0}")]
    #[allow(non_camel_case_types)]
    Required_Field(String),
    #[error("Unable to format stacktrace element: {0}")]
    Element(#[from] std::fmt::Error),
}

fn to_stacktrace_string(msg: String, backtrace_string: String) -> Result<String, StacktraceError> {
    let mut res = String::new();
    write!(&mut res, "{msg}").map_err(|error| StacktraceError::Message(error.to_string()))?;

    // Use multi-line mode and named capture groups to identify the following stacktrace fields:
    // - dc = declaredClass
    // - mn = methodName
    // - fn = fileName (optional)
    // - line = file line number (optional)
    // - col = file col number within the line (optional)
    let re = Regex::new(
        r"(?m)^\s*\d+: (?<dc>.*?)(?<mn>[^:]+)\n(\s*at\s+(?<fn>[^:]+):(?<line>\d+):(?<col>\d+)$)?",
    )?;
    for c in re.captures_iter(backtrace_string.as_str()) {
        write!(
            &mut res,
            "\n        at {}{}({}:{})",
            c.name("dc")
                .ok_or_else(|| StacktraceError::Required_Field("declared class".to_string()))?
                .as_str(),
            c.name("mn")
                .ok_or_else(|| StacktraceError::Required_Field("method name".to_string()))?
                .as_str(),
            // There are internal calls within the backtrace that don't provide file information
            c.name("fn").map(|m| m.as_str()).unwrap_or("__internal__"),
            c.name("line")
                .map(|m| m.as_str().parse().expect("numeric line number"))
                .unwrap_or(0)
        )?;
    }

    Ok(res)
}

// It is currently undefined behavior to unwind from Rust code into foreign code, so we can wrap
// our JNI functions and turn these panics into a `RuntimeException`.
pub fn try_unwrap_or_throw<T, F>(env: &EnvUnowned, f: F) -> T
where
    T: JNIDefault,
    F: FnOnce(&mut Env) -> Result<T, CometError> + UnwindSafe,
{
    let raw = env.as_raw();
    let mut env1 = unsafe { EnvUnowned::from_raw(raw) };
    match env1.with_env(f).into_outcome() {
        Outcome::Ok(value) => value,
        Outcome::Err(err) => {
            let mut guard = unsafe { jni::AttachGuard::from_unowned(raw) };
            unwrap_or_throw_default(guard.borrow_env_mut(), Err(err))
        }
        Outcome::Panic(payload) => {
            let mut guard = unsafe { jni::AttachGuard::from_unowned(raw) };
            unwrap_or_throw_default(guard.borrow_env_mut(), Err(CometError::from(payload)))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::{
        fs::File,
        io,
        io::Read,
        path::PathBuf,
        sync::{Arc, Once},
    };

    use jni::{
        objects::{JClass, JIntArray, JString, JThrowable},
        sys::{jintArray, jstring},
        EnvUnowned, InitArgsBuilder, JNIVersion, JavaVM,
    };

    use assertables::assert_starts_with;

    pub fn jvm() -> &'static Arc<JavaVM> {
        static mut JVM: Option<Arc<JavaVM>> = None;
        static INIT: Once = Once::new();

        // Capture panic backtraces
        init();

        INIT.call_once(|| {
            // Add comet-common classes to the classpath so we can find the Comet exception
            // classes (CometNativeException, CometQueryExecutionException, etc.).
            let mut common_classes = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
            common_classes.push("../../common/target/classes");
            let mut class_path = common_classes
                .as_path()
                .to_str()
                .expect("common classes as an str")
                .to_string();
            class_path.insert_str(0, "-Djava.class.path=");

            // Build the VM properties
            let jvm_args = InitArgsBuilder::new()
                // Pass the JNI API version (default is 8)
                .version(JNIVersion::V1_8)
                // You can additionally pass any JVM options (standard, like a system property,
                // or VM-specific).
                // Here we enable some extra JNI checks useful during development
                .option("-Xcheck:jni")
                .option(class_path.as_str())
                .build()
                .unwrap_or_else(|e| panic!("{e:#?}"));

            let jvm = JavaVM::new(jvm_args).unwrap_or_else(|e| panic!("{e:#?}"));

            #[allow(static_mut_refs)]
            unsafe {
                JVM = Some(Arc::new(jvm));
            }
        });

        #[allow(static_mut_refs)]
        unsafe {
            JVM.as_ref().unwrap()
        }
    }

    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn error_from_panic() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                let env_unowned = unsafe { EnvUnowned::from_raw(env.get_raw()) };
                try_unwrap_or_throw(&env_unowned, |_| -> CometResult<()> {
                    panic!("oops!");
                });

                assert_pending_java_exception_detailed(
                    env,
                    Some("java/lang/RuntimeException"),
                    Some("oops!"),
                );
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an object are handled correctly.  This is basically
    // a test of the "happy path".
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn object_result() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                let clazz = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let input = env.new_string("World").unwrap();

                let actual = unsafe {
                    Java_Errors_hello(&EnvUnowned::from_raw(env.get_raw()), clazz, input)
                };
                let actual_s = unsafe { JString::from_raw(env, actual) };

                let actual_string = actual_s.try_to_string(env).unwrap();
                assert_eq!("Hello, World!", actual_string);
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an native time are handled correctly.  This is basically
    // a test of the "happy path".
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn jlong_result() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                // Class java.lang.object is just a stand-in
                let class = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let a: jlong = 6;
                let b: jlong = 3;
                let actual =
                    unsafe { Java_Errors_div(&EnvUnowned::from_raw(env.get_raw()), class, a, b) };

                assert_eq!(2, actual);
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an array can handle throwing exceptions.  The test
    // causes an exception by dividing by zero.
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn jlong_panic_exception() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                // Class java.lang.object is just a stand-in
                let class = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let a: jlong = 6;
                let b: jlong = 0;
                let _actual =
                    unsafe { Java_Errors_div(&EnvUnowned::from_raw(env.get_raw()), class, a, b) };

                assert_pending_java_exception_detailed(
                    env,
                    Some("java/lang/RuntimeException"),
                    Some("attempt to divide by zero"),
                );
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an native time are handled correctly.  This is basically
    // a test of the "happy path".
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn jlong_result_ok() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                // Class java.lang.object is just a stand-in
                let class = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let a: JString = env.new_string("9").unwrap();
                let b: JString = env.new_string("3").unwrap();
                let actual = unsafe {
                    Java_Errors_div_with_parse(&EnvUnowned::from_raw(env.get_raw()), class, a, b)
                };

                assert_eq!(3, actual);
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an native time are handled correctly.  This is basically
    // a test of the "happy path".
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn jlong_result_err() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                // Class java.lang.object is just a stand-in
                let class = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let a: JString = env.new_string("NaN").unwrap();
                let b: JString = env.new_string("3").unwrap();
                let _actual = unsafe {
                    Java_Errors_div_with_parse(&EnvUnowned::from_raw(env.get_raw()), class, a, b)
                };

                assert_pending_java_exception_detailed(
                    env,
                    Some("java/lang/NumberFormatException"),
                    Some("invalid digit found in string"),
                );
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an array are handled correctly.  This is basically
    // a test of the "happy path".
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn jint_array_result() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                // Class java.lang.object is just a stand-in
                let class = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let buf = [2, 4, 6];
                let input = env.new_int_array(3).unwrap();
                input.set_region(env, 0, &buf).unwrap();
                let actual = unsafe {
                    Java_Errors_array_div(&EnvUnowned::from_raw(env.get_raw()), class, &input, 2)
                };
                let actual_s = unsafe { JIntArray::from_raw(env, actual) };

                let mut buf: [i32; 3] = [0; 3];
                actual_s.get_region(env, 0, &mut buf).unwrap();
                assert_eq!([1, 2, 3], buf);
                Ok(())
            })
            .unwrap();
    }

    // Verify that functions that return an array can handle throwing exceptions.  The test
    // causes an exception by dividing by zero.
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn jint_array_panic_exception() {
        jvm()
            .attach_current_thread(|env| -> jni::errors::Result<()> {
                // Class java.lang.object is just a stand-in
                let class = env.find_class(jni::jni_str!("java/lang/Object")).unwrap();
                let buf = [2, 4, 6];
                let input = env.new_int_array(3).unwrap();
                input.set_region(env, 0, &buf).unwrap();
                let _actual = unsafe {
                    Java_Errors_array_div(&EnvUnowned::from_raw(env.get_raw()), class, &input, 0)
                };

                assert_pending_java_exception_detailed(
                    env,
                    Some("java/lang/RuntimeException"),
                    Some("attempt to divide by zero"),
                );
                Ok(())
            })
            .unwrap();
    }

    /// Test that conversion of a serialized backtrace to an equivalent stacktrace message.
    ///
    /// See [`object_panic_exception`] for a test which involves generating a panic and verifying
    /// that the resulting stack trace includes the offending call.
    #[test]
    #[cfg_attr(miri, ignore)] // miri can't call foreign function `dlopen`
    pub fn stacktrace_string() {
        // Setup: Start with a backtrace that includes all of the expected scenarios, including
        // cases where the file and location are not provided as part of the backtrace capture
        let backtrace_string = read_resource("testdata/backtrace.txt").expect("backtrace content");

        // Test: Reformat the serialized backtrace as a multi-line message which includes the
        // backtrace formatted as a stacktrace
        let stacktrace_string =
            to_stacktrace_string("Some Error Message".to_string(), backtrace_string).unwrap();

        // Verify: The message matches the expected output.  Trim the expected string to remove
        // the carriage return
        let expected_string = read_resource("testdata/stacktrace.txt").expect("stacktrace content");
        assert_eq!(expected_string.trim(), stacktrace_string.as_str());
    }

    fn read_resource(path: &str) -> Result<String, io::Error> {
        let mut path_buf = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        path_buf.push(path);

        let mut f = File::open(path_buf.as_path())?;
        let mut s = String::new();
        f.read_to_string(&mut s)?;
        Ok(s)
    }

    // Example of a simple JNI "Hello World" program.  It can be used to demonstrate:
    // * returning an object
    // * throwing an exception from `.expect()`
    #[no_mangle]
    pub extern "system" fn Java_Errors_hello(
        e: &EnvUnowned,
        _class: JClass,
        input: JString,
    ) -> jstring {
        try_unwrap_or_throw(e, |env| {
            let input: String = input.try_to_string(env).expect("Couldn't get java string!");

            let output = env
                .new_string(format!("Hello, {input}!"))
                .expect("Couldn't create java string!");

            Ok(output.into_raw())
        })
    }

    // Example of a simple JNI function that divides.  It can be used to demonstrate:
    // * returning an native type
    // * throwing an exception when dividing by zero
    #[no_mangle]
    pub extern "system" fn Java_Errors_div(
        env: &EnvUnowned,
        _class: JClass,
        a: jlong,
        b: jlong,
    ) -> jlong {
        try_unwrap_or_throw(env, |_| Ok(a / b))
    }

    #[no_mangle]
    pub extern "system" fn Java_Errors_div_with_parse(
        e: &EnvUnowned,
        _class: JClass,
        a: JString,
        b: JString,
    ) -> jlong {
        try_unwrap_or_throw(e, |env| {
            let a_value: i64 = a.try_to_string(env)?.parse()?;
            let b_value: i64 = b.try_to_string(env)?.parse()?;
            Ok(a_value / b_value)
        })
    }

    // Example of a simple JNI function that divides.  It can be used to demonstrate:
    // * returning an array
    // * throwing an exception when dividing by zero
    #[no_mangle]
    pub extern "system" fn Java_Errors_array_div(
        e: &EnvUnowned,
        _class: JClass,
        input: &JIntArray,
        divisor: jint,
    ) -> jintArray {
        try_unwrap_or_throw(e, |env| {
            let mut input_buf: [jint; 3] = [0; 3];
            input.get_region(env, 0, &mut input_buf)?;

            let buf = input_buf.map(|v| -> jint { v / divisor });

            let result = env.new_int_array(3)?;
            result.set_region(env, 0, &buf)?;
            Ok(result.into_raw())
        })
    }

    // Helper method that asserts there is a pending Java exception which is an `instance_of`
    // `expected_type` with a message matching `expected_message` and clears it if any.
    fn assert_pending_java_exception_detailed(
        env: &mut Env,
        expected_type: Option<&str>,
        expected_message: Option<&str>,
    ) {
        assert!(env.exception_check());
        let exception = env.exception_occurred().expect("Unable to get exception");
        env.exception_clear();

        if let Some(expected_type) = expected_type {
            assert_exception_type(env, &exception, expected_type);
        }

        if let Some(expected_message) = expected_message {
            assert_exception_message(env, exception, expected_message);
        }
    }

    // Asserts that exception is an `instance_of` `expected_type` type.
    fn assert_exception_type(env: &mut Env, exception: &JThrowable, expected_type: &str) {
        if !env
            .is_instance_of(exception, jni::strings::JNIString::new(expected_type))
            .unwrap()
        {
            let class: JClass = env.get_object_class(exception).unwrap();
            let name = env
                .call_method(
                    class,
                    jni::jni_str!("getName"),
                    jni::jni_sig!("()Ljava/lang/String;"),
                    &[],
                )
                .unwrap()
                .l()
                .unwrap();
            let name_string = unsafe { JString::from_raw(env, name.into_raw()) };
            let class_name: String = name_string.try_to_string(env).unwrap();
            assert_eq!(class_name.replace('.', "/"), expected_type);
        };
    }

    // Asserts that exception's message matches `expected_message`.
    fn assert_exception_message(env: &mut Env, exception: JThrowable, expected_message: &str) {
        let message = env
            .call_method(
                exception,
                jni::jni_str!("getMessage"),
                jni::jni_sig!("()Ljava/lang/String;"),
                &[],
            )
            .unwrap()
            .l()
            .unwrap();
        let message_string = unsafe { JString::from_raw(env, message.into_raw()) };
        let msg_rust: String = message_string.try_to_string(env).unwrap();
        println!("{msg_rust}");
        // Since panics result in multi-line messages which include the backtrace, just use the
        // first line.
        assert_starts_with!(msg_rust, expected_message);
    }

    // --- try_classify_file_read_error: typed classification of file-read DataFusionErrors ---
    // These guard the variant matching that replaced JVM-side error-message string matching. They
    // need no JVM (pure DataFusionError -> Option<SparkError>), so they also run under miri.

    use datafusion::common::DataFusionError;

    fn file_path_of(err: &SparkError) -> &str {
        match err {
            SparkError::CannotReadFile { file_path, .. } => file_path,
            other => panic!("expected CannotReadFile, got {other:?}"),
        }
    }

    #[test]
    fn classify_parquet_error_is_file_read() {
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General(
            "corrupt footer".to_string(),
        )));
        let classified = try_classify_file_read_error(&e);
        assert!(
            classified.is_some(),
            "ParquetError should classify as file-read"
        );
        // No path available from a bare ParquetError; JVM fills it from the per-task list.
        assert_eq!(file_path_of(&classified.unwrap()), "");
    }

    #[test]
    fn classify_arrow_parquet_error_is_file_read() {
        // arrow-rs surfaces some parquet read failures as ArrowError::ParquetError.
        let e = DataFusionError::ArrowError(
            Box::new(ArrowError::ParquetError(
                "failed to fill whole buffer".to_string(),
            )),
            None,
        );
        assert!(
            try_classify_file_read_error(&e).is_some(),
            "ArrowError(ParquetError) should classify as file-read"
        );
    }

    #[test]
    fn classify_object_store_not_found_is_file_not_found() {
        // A genuinely-missing file must classify as FileNotFound (-> readCurrentFileNotFoundError
        // on the JVM side), NOT CannotReadFile (-> cannotReadFilesError). The path is carried in
        // the message; the JVM shim extracts it.
        let e = DataFusionError::ObjectStore(Box::new(datafusion::object_store::Error::NotFound {
            path: "file:/tmp/data/part-3.parquet".to_string(),
            source: "missing".into(),
        }));
        match try_classify_file_read_error(&e) {
            Some(SparkError::FileNotFound { message }) => {
                assert!(message.contains("part-3.parquet"), "message was: {message}")
            }
            other => panic!("expected FileNotFound, got {other:?}"),
        }
    }

    #[test]
    fn classify_parquet_error_wrapping_not_found_is_file_not_found() {
        // The parquet reader surfaces a missing data file as ParquetError::External wrapping an
        // object_store NotFound (e.g. a Delta CDC file removed by VACUUM). The NotFound is only
        // reachable through the source chain, but it must still classify as FileNotFound.
        let os = datafusion::object_store::Error::NotFound {
            path: "file:/tmp/t/_change_data/cdc.parquet".to_string(),
            source: "missing".into(),
        };
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::External(
            Box::new(os),
        )));
        match try_classify_file_read_error(&e) {
            Some(SparkError::FileNotFound { .. }) => {}
            other => {
                panic!("expected FileNotFound for a NotFound-wrapping ParquetError, got {other:?}")
            }
        }
    }

    #[test]
    fn classify_flattened_metadata_not_found_is_file_not_found() {
        // DataFusion's CachedParquetFileReader flattens NotFound into a sourceless General error;
        // classification falls back to the message. See #3978.
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General(
            "Failed to fetch metadata for file /tmp/t/missing.parquet: Object Store error: \
             Object at location /tmp/t/missing.parquet not found: No such file or directory \
             (os error 2)"
                .to_string(),
        )));
        match try_classify_file_read_error(&e) {
            Some(SparkError::FileNotFound { .. }) => {}
            other => {
                panic!("expected FileNotFound for a flattened metadata NotFound, got {other:?}")
            }
        }
    }

    #[test]
    fn classify_corrupt_parquet_error_stays_cannot_read_file() {
        // A corrupt/truncated file (no NotFound in the chain) must remain CannotReadFile
        // (-> FAILED_READ_FILE), unchanged by the FileNotFound carve-out.
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General(
            "corrupt footer".to_string(),
        )));
        match try_classify_file_read_error(&e) {
            Some(SparkError::CannotReadFile { .. }) => {}
            other => panic!("expected CannotReadFile for a corrupt parquet error, got {other:?}"),
        }
    }

    #[test]
    fn classify_pushed_filter_predicate_error_is_not_file_read() {
        // A pushed-down WHERE predicate that hits an ANSI divide-by-zero while the parquet scan is
        // reading is surfaced by DataFusion's row filter as an ArrowError (ComputeError), which the
        // parquet reader wraps as ParquetError::External(<arrow>). That is an expression failure,
        // NOT a file read, so it must NOT classify as CannotReadFile/FAILED_READ_FILE -- it must
        // fall through (None) so the underlying error surfaces verbatim.
        let predicate_err = ArrowError::ComputeError(
            "Error evaluating filter predicate: External(SparkErrorWithContext { \
             error: DivideByZero, context: Some(..) })"
                .to_string(),
        );
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::External(
            Box::new(predicate_err),
        )));
        assert!(
            try_classify_file_read_error(&e).is_none(),
            "a pushed-down filter-predicate eval error must not be classified as a file read; got {:?}",
            try_classify_file_read_error(&e)
        );
    }

    #[test]
    fn classify_external_io_parquet_error_stays_cannot_read_file() {
        // The structural carve-out above must be narrow: a ParquetError::External wrapping a genuine
        // IO error (a real read failure, not an ArrowError) must STILL classify as CannotReadFile.
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::External(
            Box::new(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                "early eof reading footer",
            )),
        )));
        match try_classify_file_read_error(&e) {
            Some(SparkError::CannotReadFile { .. }) => {}
            other => {
                panic!("expected CannotReadFile for an External(io) parquet error, got {other:?}")
            }
        }
    }

    #[test]
    fn classify_invalid_parquet_file_carries_spark_wording() {
        // parquet-rs reports a bad magic / unreadable footer as "Invalid Parquet file. Corrupt
        // footer"; Spark's ParquetQuerySuite asserts the cause says "is not a Parquet file". The
        // CannotReadFile message must carry that phrasing.
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General(
            "Invalid Parquet file. Corrupt footer".to_string(),
        )));
        match try_classify_file_read_error(&e) {
            Some(SparkError::CannotReadFile { message, .. }) => assert!(
                message.contains("is not a Parquet file"),
                "message was: {message}"
            ),
            other => panic!("expected CannotReadFile, got {other:?}"),
        }
    }

    #[test]
    fn classify_other_parquet_error_keeps_native_message() {
        // A non-magic parquet failure (e.g. corrupt page data) keeps its native message verbatim
        // -- the "is not a Parquet file" phrasing is only added for the magic/footer case.
        let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General(
            "could not decode page header".to_string(),
        )));
        match try_classify_file_read_error(&e) {
            Some(SparkError::CannotReadFile { message, .. }) => assert!(
                !message.contains("is not a Parquet file"),
                "message should not be augmented: {message}"
            ),
            other => panic!("expected CannotReadFile, got {other:?}"),
        }
    }

    #[test]
    fn classify_bare_io_error_is_not_file_read() {
        // A bare IoError is not a scan read failure (scans surface ParquetError/ObjectStore); it
        // can come from spill/shuffle, so it must NOT be classified as FAILED_READ_FILE.
        let e = DataFusionError::IoError(io::Error::new(io::ErrorKind::UnexpectedEof, "eof"));
        assert!(try_classify_file_read_error(&e).is_none());
    }

    #[test]
    fn classify_unwraps_context_and_shared() {
        let inner = DataFusionError::ParquetError(Box::new(
            parquet::errors::ParquetError::General("corrupt".to_string()),
        ));
        let ctx = DataFusionError::Context("reading file".to_string(), Box::new(inner));
        assert!(
            try_classify_file_read_error(&ctx).is_some(),
            "Context-wrapped ParquetError should classify"
        );
        let shared = DataFusionError::Shared(Arc::new(DataFusionError::ObjectStore(Box::new(
            datafusion::object_store::Error::NotFound {
                path: "p".to_string(),
                source: "x".into(),
            },
        ))));
        assert!(
            try_classify_file_read_error(&shared).is_some(),
            "Shared-wrapped ObjectStore error should classify"
        );
    }

    #[test]
    fn classify_non_file_errors_are_not_file_read() {
        // Execution / Internal errors (and object_store Generic config errors, which arrive as
        // Execution strings) must NOT be masked as file-read failures.
        assert!(try_classify_file_read_error(&DataFusionError::Execution(
            "Hdfs support is not enabled in this build".to_string()
        ))
        .is_none());
        assert!(
            try_classify_file_read_error(&DataFusionError::Internal("bug".to_string())).is_none()
        );
    }
}