refactor: extract spark-errors crate to replace callback pattern

Move SparkError, SparkErrorWithContext, QueryContext, and QueryContextMap
into a new spark-errors crate that both jvm-bridge and spark-expr depend
on. This allows jvm-bridge to directly downcast SparkError variants in
throw_exception, eliminating the ExternalErrorHandler callback, OnceCell,
and init-time registration that were previously needed.

Author: andygrove

native/Cargo.lock

@@ -16,8 +16,8 @@
 # under the License.
 
 [workspace]
-default-members = ["core", "spark-expr", "proto", "jvm-bridge"]
-members = ["core", "spark-expr", "proto", "jvm-bridge", "hdfs", "fs-hdfs"]
+default-members = ["core", "spark-expr", "spark-errors", "proto", "jvm-bridge"]
+members = ["core", "spark-expr", "spark-errors", "proto", "jvm-bridge", "hdfs", "fs-hdfs"]
 resolver = "2"
 
 [workspace.package]
@@ -43,6 +43,7 @@ datafusion-datasource = { version = "52.2.0" }
 datafusion-physical-expr-adapter = { version = "52.2.0" }
 datafusion-spark = { version = "52.2.0" }
 datafusion-comet-spark-expr = { path = "spark-expr" }
+datafusion-comet-spark-errors = { path = "spark-errors" }
 datafusion-comet-jvm-bridge = { path = "jvm-bridge" }
 datafusion-comet-proto = { path = "proto" }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }

native/Cargo.toml

@@ -91,9 +91,6 @@ pub extern "system" fn Java_org_apache_comet_NativeBase_init(
     // Initialize the error handling to capture panic backtraces
     errors::init();
 
-    // Register SparkError handler for JNI exception throwing
-    errors::register_external_error_handler(handle_spark_error);
-
     try_unwrap_or_throw(&e, |mut env| {
         let path: String = env.get_string(&log_conf_path)?.into();
 
@@ -122,33 +119,6 @@ pub extern "system" fn Java_org_apache_comet_NativeBase_init(
     })
 }
 
-/// Handle SparkError variants in DataFusionError::External for JNI exception throwing.
-/// Returns true if the error was handled.
-fn handle_spark_error(
-    error: &(dyn std::error::Error + Send + Sync + 'static),
-    env: &mut JNIEnv,
-) -> bool {
-    use datafusion_comet_spark_expr::{SparkError, SparkErrorWithContext};
-
-    if let Some(spark_error_with_ctx) = error.downcast_ref::<SparkErrorWithContext>() {
-        let json_message = spark_error_with_ctx.to_json();
-        let _ = env.throw_new(
-            "org/apache/comet/exceptions/CometQueryExecutionException",
-            json_message,
-        );
-        return true;
-    }
-    if let Some(spark_error) = error.downcast_ref::<SparkError>() {
-        let json_message = spark_error.to_json();
-        let _ = env.throw_new(
-            "org/apache/comet/exceptions/CometQueryExecutionException",
-            json_message,
-        );
-        return true;
-    }
-    false
-}
-
 const LOG_PATTERN: &str = "{d(%y/%m/%d %H:%M:%S)} {l} {f}: {m}{n}";
 
 /// JNI method to check if a specific feature is enabled in the native Rust code.

native/core/src/lib.rs

@@ -39,6 +39,7 @@ lazy_static = "1.4.0"
 once_cell = "1.18.0"
 paste = "1.0.14"
 prost = "0.14.3"
+datafusion-comet-spark-errors = { workspace = true }
 
 [dev-dependencies]
 jni = { version = "0.21", features = ["invocation"] }

native/jvm-bridge/Cargo.toml

@@ -19,6 +19,7 @@
 
 use arrow::error::ArrowError;
 use datafusion::common::DataFusionError;
+use datafusion_comet_spark_errors::{SparkError, SparkErrorWithContext};
 use jni::errors::{Exception, ToException};
 use regex::Regex;
 
@@ -40,25 +41,9 @@ use jni::sys::{jboolean, jbyte, jchar, jdouble, jfloat, jint, jlong, jobject, js
 use jni::objects::{GlobalRef, JThrowable};
 use jni::JNIEnv;
 use lazy_static::lazy_static;
-use once_cell::sync::OnceCell;
 use parquet::errors::ParquetError;
 use thiserror::Error;
 
-/// Handler for DataFusionError::External errors during JNI exception throwing.
-/// Returns true if the error was handled (exception thrown), false to fall through
-/// to the default handler.
-pub type ExternalErrorHandler =
-    fn(error: &(dyn std::error::Error + Send + Sync + 'static), env: &mut JNIEnv) -> bool;
-
-static EXTERNAL_ERROR_HANDLER: OnceCell<ExternalErrorHandler> = OnceCell::new();
-
-/// Register a handler for DataFusionError::External errors.
-/// This allows the core crate to register SparkError-specific handling
-/// without the jvm-bridge crate needing to know about SparkError.
-pub fn register_external_error_handler(handler: ExternalErrorHandler) {
-    let _ = EXTERNAL_ERROR_HANDLER.set(handler);
-}
-
 lazy_static! {
     static ref PANIC_BACKTRACE: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
 }
@@ -489,26 +474,34 @@ fn throw_exception(env: &mut JNIEnv, error: &CometError, backtrace: Option<Strin
                         throwable,
                     },
             } => env.throw(<&JThrowable>::from(throwable.as_obj())),
-            // Handle DataFusion errors containing external errors (e.g. SparkError)
+            // Handle DataFusion errors containing SparkError or SparkErrorWithContext
             CometError::DataFusion {
                 msg: _,
                 source: DataFusionError::External(e),
             } => {
-                // Try registered handler first (e.g. SparkError handling from core)
-                if let Some(handler) = EXTERNAL_ERROR_HANDLER.get() {
-                    if handler(e.as_ref(), env) {
-                        return;
+                if let Some(spark_error_with_ctx) = e.downcast_ref::<SparkErrorWithContext>() {
+                    let json_message = spark_error_with_ctx.to_json();
+                    env.throw_new(
+                        "org/apache/comet/exceptions/CometQueryExecutionException",
+                        json_message,
+                    )
+                } else if let Some(spark_error) = e.downcast_ref::<SparkError>() {
+                    let json_message = spark_error.to_json();
+                    env.throw_new(
+                        "org/apache/comet/exceptions/CometQueryExecutionException",
+                        json_message,
+                    )
+                } else {
+                    // Fall through to generic exception
+                    let exception = error.to_exception();
+                    match backtrace {
+                        Some(backtrace_string) => env.throw_new(
+                            exception.class,
+                            to_stacktrace_string(exception.msg, backtrace_string).unwrap(),
+                        ),
+                        _ => env.throw_new(exception.class, exception.msg),
                     }
                 }
-                // Fall through to generic exception
-                let exception = error.to_exception();
-                match backtrace {
-                    Some(backtrace_string) => env.throw_new(
-                        exception.class,
-                        to_stacktrace_string(exception.msg, backtrace_string).unwrap(),
-                    ),
-                    _ => env.throw_new(exception.class, exception.msg),
-                }
             }
             _ => {
                 let exception = error.to_exception();

native/jvm-bridge/src/errors.rs

@@ -0,0 +1,40 @@
+# 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.
+
+[package]
+name = "datafusion-comet-spark-errors"
+description = "Apache DataFusion Comet: Spark error types"
+version = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+authors = { workspace = true }
+readme = { workspace = true }
+license = { workspace = true }
+edition = { workspace = true }
+
+publish = false
+
+[dependencies]
+arrow = { workspace = true }
+datafusion = { workspace = true }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+thiserror = { workspace = true }
+
+[lib]
+name = "datafusion_comet_spark_errors"
+path = "src/lib.rs"