fix: tolerate missing CometUdfBridge class at JVMClasses init

Make comet_udf_bridge an Option in JVMClasses so a missing
org.apache.comet.udf.CometUdfBridge class (e.g. shading dropped
org.apache.comet.udf.*) no longer crashes executor JVM init. The
JVM-UDF dispatch path returns a clear ExecutionError when the bridge
is unavailable. Also clarify the FFI lifetime contract on the result
import.

Author: andygrove

native/jni-bridge/src/lib.rs

@@ -220,7 +220,9 @@ pub struct JVMClasses<'a> {
     /// acquire & release native memory.
     pub comet_task_memory_manager: CometTaskMemoryManager<'a>,
     /// The CometUdfBridge class used to dispatch JVM scalar UDFs.
-    pub comet_udf_bridge: CometUdfBridge<'a>,
+    /// `None` if the class is not on the classpath; the JVM-UDF dispatch path
+    /// reports a clear error rather than crashing executor init.
+    pub comet_udf_bridge: Option<CometUdfBridge<'a>>,
 }
 
 unsafe impl Send for JVMClasses<'_> {}
@@ -291,7 +293,16 @@ impl JVMClasses<'_> {
                 comet_batch_iterator: CometBatchIterator::new(env).unwrap(),
                 comet_shuffle_block_iterator: CometShuffleBlockIterator::new(env).unwrap(),
                 comet_task_memory_manager: CometTaskMemoryManager::new(env).unwrap(),
-                comet_udf_bridge: CometUdfBridge::new(env).unwrap(),
+                comet_udf_bridge: {
+                    // Optional: if the bridge class is absent (e.g. comet shading
+                    // dropped org.apache.comet.udf.*), record None and clear the
+                    // pending JVM exception so other JNI calls keep working.
+                    let bridge = CometUdfBridge::new(env).ok();
+                    if env.exception_check() {
+                        env.exception_clear();
+                    }
+                    bridge
+                },
             }
         });
     }

native/spark-expr/src/jvm_udf/mod.rs

@@ -29,7 +29,7 @@ use datafusion::common::Result as DFResult;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 
-use datafusion_comet_jni_bridge::errors::CometError;
+use datafusion_comet_jni_bridge::errors::{CometError, ExecutionError};
 use datafusion_comet_jni_bridge::JVMClasses;
 use jni::objects::{JObject, JValue};
 
@@ -148,7 +148,14 @@ impl PhysicalExpr for JvmScalarUdfExpr {
 
         // Step 3: attach a JNI env for this thread and call the static bridge method.
         JVMClasses::with_env(|env| {
-            let bridge = &JVMClasses::get().comet_udf_bridge;
+            let bridge = JVMClasses::get().comet_udf_bridge.as_ref().ok_or_else(|| {
+                CometError::from(ExecutionError::GeneralError(
+                    "JVM UDF bridge unavailable: org.apache.comet.udf.CometUdfBridge \
+                     class was not found on the JVM classpath. Set \
+                     spark.comet.exec.regexp.engine=rust to disable this path."
+                        .to_string(),
+                ))
+            })?;
 
             // Build the JVM String for the class name.
             let jclass_name = env
@@ -195,8 +202,11 @@ impl PhysicalExpr for JvmScalarUdfExpr {
         })?;
 
         // Step 4: import the result from the FFI slots filled by the JVM.
-        // SAFETY: from_ffi consumes the FFI_ArrowArray's release ownership; arrow-rs sets the
-        // release callback to None on the moved-from struct, so the Box's subsequent drop is a no-op.
+        // SAFETY: `*out_array` moves the FFI_ArrowArray out of the Box (the heap
+        // allocation is freed by the move), and `from_ffi` wraps it in an Arc that
+        // keeps the JVM-installed release callback alive until the resulting
+        // ArrayData drops. `out_schema` is borrowed; its release callback runs
+        // exactly once when the Box drops at end of scope.
         let result_data = unsafe { from_ffi(*out_array, &out_schema) }
             .map_err(|e| CometError::Arrow { source: e })?;
         Ok(ColumnarValue::Array(make_array(result_data)))