refactor: route regex expressions through codegen dispatcher instead of hand-written UDFs

Replace the six hand-written `RegExp*UDF` / `StringSplitUDF` JVM UDF
implementations with the Arrow-direct codegen dispatcher introduced in
PR apache#4417 (`CometScalaUDF.emitJvmCodegenDispatch`). The dispatcher
Janino-compiles Spark's own `doGenCode` for the expression, so the
regex family inherits Spark-identical semantics with no per-expression
glue code.

Changes:

- Delete `spark/src/main/scala/org/apache/comet/udf/RegExp*UDF.scala`
  and `StringSplitUDF.scala`. Their behavior is now provided by
  Spark's `doGenCode` running inside the dispatcher.
- Rewrite the regex serdes in `strings.scala`. Expressions with no
  native Rust path (`RegExpExtract`, `RegExpExtractAll`, `RegExpInStr`)
  share a new `CometRegexpCodegenOnly` base; expressions with a native
  path (`RLike`, `RegExpReplace`, `StringSplit`) keep an explicit
  route table where the JVM arm now delegates to
  `CometScalaUDF.emitJvmCodegenDispatch`.
- Drop the `spark.comet.jvmUdf.enabled` config. The codegen dispatcher
  already has its own master switch
  (`spark.comet.exec.scalaUDF.codegen.enabled`); gating the regex
  family on the same flag avoids two flags for the same path.
  `spark.comet.exec.regexp.engine` keeps the `java`/`rust` selector
  semantics, and `engine=java` now requires the codegen flag.
- Revert the native Rust additions in `jvm_udf/mod.rs` and
  `jni-bridge/src/lib.rs`. The codegen dispatcher constructs Arrow
  output fields JVM-side via `CometBatchKernelCodegenOutput.toFfiArrowField`,
  so the list-vector field-name normalization cast is unnecessary.
- Update `CometRegExpJvmSuite`, `CometRegExpBenchmark`, the regex SQL
  test fixtures, and the regex compatibility doc to reflect the new
  gating.

Test plan:
- `CometRegExpJvmSuite`: 45/45 pass (covers all six regex expressions
  through the codegen dispatcher).
- `CometSqlFileTestSuite`: 289/289 pass.
- `CometStringExpressionSuite`: 33/33 pass.
- `CometCodegenSuite`: 60/60 pass.
- `cargo clippy --all-targets --workspace -- -D warnings`: clean.

Author: andygrove

docs/source/user-guide/latest/compatibility/regex.md

@@ -20,19 +20,20 @@ under the License.
 # Regular Expressions
 
 Comet provides two regexp engines for evaluating regular expressions: a **Rust engine** that uses the Rust
-[`regex`] crate natively, and an experimental **Java engine** that calls back into the JVM via Comet's JVM
-UDF framework. The engine is selected with `spark.comet.exec.regexp.engine`, which accepts:
+[`regex`] crate natively, and an experimental **Java engine** that runs Spark's own `doGenCode` for the
+expression inside Comet's Arrow-direct codegen dispatcher (the same dispatcher used by Comet's
+`ScalaUDF` codegen path). The engine is selected with `spark.comet.exec.regexp.engine`, which accepts:
 
 - `java` (default) — route through the Java engine for full Spark compatibility. Requires
-  `spark.comet.jvmUdf.enabled=true`; otherwise regex expressions fall back to Spark with an explanatory
-  message.
+  `spark.comet.exec.scalaUDF.codegen.enabled=true`; otherwise regex expressions fall back to Spark with
+  an explanatory message.
 - `rust` — run the Rust engine when an expression has a native implementation. Setting this is itself
   the opt-in for the semantic differences between Java and Rust regex (no separate `allowIncompatible`
   flag needed). Expressions without a native Rust implementation (`regexp_extract`,
   `regexp_extract_all`, `regexp_instr`) fall back to Spark.
 
-The JVM UDF framework is experimental and disabled by default. With pure defaults
-(`engine=java`, `jvmUdf.enabled=false`), all regex expressions fall back to Spark.
+The codegen dispatcher is experimental and disabled by default. With pure defaults
+(`engine=java`, `scalaUDF.codegen.enabled=false`), all regex expressions fall back to Spark.
 
 ## Choosing an engine
 
@@ -47,9 +48,9 @@ The **Rust engine** is faster but cannot match Java regex semantics for every pa
 choice is itself the opt-in, setting `spark.comet.exec.regexp.engine=rust` declares acceptance of those
 differences without a separate per-expression flag.
 
-The **Java engine** is the default but the underlying JVM UDF framework is experimental and gated behind
-`spark.comet.jvmUdf.enabled=true`; the behavior, configuration, and supported expressions may change in
-future releases.
+The **Java engine** is the default but the underlying codegen dispatcher is experimental and gated behind
+`spark.comet.exec.scalaUDF.codegen.enabled=true`; the behavior, configuration, and supported expressions
+may change in future releases.
 
 ## Why the engines differ
 

native/jni-bridge/src/lib.rs

@@ -231,8 +231,7 @@ pub struct JVMClasses<'a> {
     /// acquire & release native memory.
     pub comet_task_memory_manager: CometTaskMemoryManager<'a>,
     /// The CometUdfBridge class used to dispatch JVM scalar UDFs.
-    /// `None` if the class is not on the classpath; the JVM-UDF dispatch path
-    /// reports a clear error rather than crashing executor init.
+    /// `None` if the class is not on the classpath.
     pub comet_udf_bridge: Option<CometUdfBridge<'a>>,
 }
 
@@ -305,9 +304,6 @@ impl JVMClasses<'_> {
                 comet_shuffle_block_iterator: CometShuffleBlockIterator::new(env).unwrap(),
                 comet_task_memory_manager: CometTaskMemoryManager::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();

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

@@ -179,8 +179,7 @@ impl PhysicalExpr for JvmScalarUdfExpr {
             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."
+                     class was not found on the JVM classpath."
                         .to_string(),
                 ))
             })?;
@@ -244,19 +243,7 @@ impl PhysicalExpr for JvmScalarUdfExpr {
         // 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 })?;
-        let result_array = make_array(result_data);
-
-        // The JVM may produce arrays with different field names (e.g. Arrow Java's
-        // ListVector uses "$data$" for child fields) than what DataFusion expects
-        // (e.g. "item"). Cast to the declared return_type to normalize schema.
-        let result_array = if result_array.data_type() != &self.return_type {
-            arrow::compute::cast(&result_array, &self.return_type)
-                .map_err(|e| CometError::Arrow { source: e })?
-        } else {
-            result_array
-        };
-
-        Ok(ColumnarValue::Array(result_array))
+        Ok(ColumnarValue::Array(make_array(result_data)))
     }
 
     fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -362,15 +362,15 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
-  val COMET_JVM_UDF_ENABLED: ConfigEntry[Boolean] =
-    conf("spark.comet.jvmUdf.enabled")
+  val COMET_SCALA_UDF_CODEGEN_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.exec.scalaUDF.codegen.enabled")
       .category(CATEGORY_EXEC)
-      .doc(
-        "Master switch for the JVM UDF framework, which lets native execution call back into " +
-          "the JVM to evaluate selected expressions for full Spark compatibility (at the cost " +
-          "of a JNI round-trip per batch). The framework is experimental and may change in " +
-          "future releases. Disabled by default; expressions that would otherwise route " +
-          "through the JVM UDF bridge fall back to native or to Spark while this is false.")
+      .doc("Experimental. Whether to route Spark `ScalaUDF` expressions through Comet's " +
+        "Arrow-direct codegen dispatcher. When enabled, a supported ScalaUDF is compiled into " +
+        "a per-batch kernel that reads and writes Arrow vectors directly from native " +
+        "execution. When disabled, plans containing a ScalaUDF fall back to Spark for the " +
+        "enclosing operator. The same dispatcher backs `spark.comet.exec.regexp.engine=java` " +
+        "so the regex family routes through it as well.")
       .booleanConf
       .createWithDefault(false)
 
@@ -382,9 +382,9 @@ object CometConf extends ShimCometConf {
       .category(CATEGORY_EXEC)
       .doc(
         "Selects the engine used to evaluate Spark regular-expression expressions. " +
-          s"`$REGEXP_ENGINE_JAVA` (default) routes through a JVM-side UDF " +
-          "(java.util.regex.Pattern) for Spark-compatible semantics, at the cost of JNI " +
-          s"roundtrips per batch; this requires ${COMET_JVM_UDF_ENABLED.key}=true and " +
+          s"`$REGEXP_ENGINE_JAVA` (default) routes through the Arrow-direct codegen dispatcher " +
+          "so Spark's own `doGenCode` (backed by `java.util.regex.Pattern`) runs inside the " +
+          s"Comet pipeline; this requires ${COMET_SCALA_UDF_CODEGEN_ENABLED.key}=true and " +
           s"falls back to Spark otherwise. `$REGEXP_ENGINE_RUST` runs the native DataFusion " +
           "regexp engine when an implementation exists; setting this is itself the opt-in " +
           "for the semantic differences between Java and Rust regex. Affected expressions: " +
@@ -396,18 +396,6 @@ object CometConf extends ShimCometConf {
       .checkValues(Set(REGEXP_ENGINE_RUST, REGEXP_ENGINE_JAVA))
       .createWithDefault(REGEXP_ENGINE_JAVA)
 
-  val COMET_SCALA_UDF_CODEGEN_ENABLED: ConfigEntry[Boolean] =
-    conf("spark.comet.exec.scalaUDF.codegen.enabled")
-      .category(CATEGORY_EXEC)
-      .doc("Experimental. Whether to route Spark `ScalaUDF` expressions through Comet's " +
-        "Arrow-direct codegen dispatcher. When enabled, a supported ScalaUDF is compiled into " +
-        "a per-batch kernel that reads and writes Arrow vectors directly from native " +
-        "execution. When disabled, plans containing a ScalaUDF fall back to Spark for the " +
-        "enclosing operator.")
-      .booleanConf
-      .createWithDefault(false)
-
-
   val COMET_EXEC_SHUFFLE_WITH_HASH_PARTITIONING_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.native.shuffle.partitioning.hash.enabled")
       .category(CATEGORY_SHUFFLE)