fix: guard against silent fail_on_error loss in scalar wiring (#5074)

.github/workflows/pr_build_linux.yml

@@ -408,6 +408,7 @@ jobs:
               org.apache.comet.CometStringDecodeSuite
               org.apache.comet.CometWidthBucketSuite
               org.apache.comet.CometUuidExpressionSuite
+              org.apache.comet.serde.CometScalarFunctionSuite
       fail-fast: false
     name: ${{ matrix.profile.name }} [${{ matrix.suite.name }}]
     runs-on: ubuntu-24.04

.github/workflows/pr_build_macos.yml

@@ -224,6 +224,7 @@ jobs:
               org.apache.comet.CometStringDecodeSuite
               org.apache.comet.CometWidthBucketSuite
               org.apache.comet.CometUuidExpressionSuite
+              org.apache.comet.serde.CometScalarFunctionSuite
 
       fail-fast: false
     name: ${{ matrix.os }}/${{ matrix.profile.name }} [${{ matrix.suite.name }}]

docs/source/contributor-guide/adding_a_new_expression.md

@@ -87,6 +87,37 @@ For simple scalar functions that map directly to a DataFusion function, you can
 classOf[Cos] -> CometScalarFunction("cos")
 ```
 
+### ANSI / `fail_on_error` constraints
+
+`CometScalarFunction` uses `scalarFunctionExprToProto`, which always serializes
+`fail_on_error=false`.
+
+On the native side, `create_comet_physical_fun` delegates to
+`create_comet_physical_fun_with_eval_mode`, whose registry catch-all **fails
+closed**. If `fail_on_error=Some(true)` reaches a UDF that can only be resolved
+from the function registry—including functions registered by
+`datafusion-spark`—Comet returns an error instead of silently ignoring the flag.
+
+Follow these rules when wiring scalar expressions:
+
+- **Do not** register a Spark expression class that exposes `failOnError`,
+  `evalMode`, `nullOnOverflow`, `ansiEnabled`, or `evalContext` using plain
+  `CometScalarFunction`. `CometScalarFunction.convert` rejects such expressions
+  by returning `None`, allowing the planner to fall back to Spark.
+
+- **Prefer name-based ANSI and try variants** so that the error semantics are
+  encoded in the function name while the proto `fail_on_error` flag remains
+  `false`. Existing examples include `parse_url` / `try_parse_url`
+  (`CometParseUrl`) and `url_decode` / `try_url_decode`
+  (`CometUrlDecodeStaticInvoke`).
+
+- If a function must receive `fail_on_error` through the proto, use
+  `scalarFunctionExprToProtoWithReturnType(..., failOnError, ...)`. The function
+  must also have a dedicated match arm in
+  `create_comet_physical_fun_with_eval_mode` that actually consumes the flag;
+  it must not fall through to registry lookup. Existing native consumers
+  include `make_decimal`, `make_date`, `make_time`, and `next_day`.
+
 #### When to set the return type explicitly
 
 `CometScalarFunction(name)` and the lower-level `scalarFunctionExprToProto(name, args)` helper both produce a protobuf `ScalarFunc` message **without** a `return_type` field. That is fine when the function name does not collide with a DataFusion built-in, or when it does collide and the Spark and DataFusion versions take the same arity and types. In that case the native planner consults DataFusion's UDF registry only to resolve the return type, then swaps in Comet's UDF for execution.

native/spark-expr/src/comet_scalar_funcs.rs

@@ -260,6 +260,17 @@ pub fn create_comet_physical_fun_with_eval_mode(
             let func = Arc::new(crate::string_funcs::spark_levenshtein);
             make_comet_scalar_udf!("levenshtein", func, without data_type)
         }
+        // Spark 4.1+ serde always sets fail_on_error=true (always-throw semantics).
+        // SparkMakeTime already throws on invalid input, so accept the flag here rather
+        // than falling through to the registry fail-closed path.
+        "make_time" => Ok(Arc::new(ScalarUDF::new_from_impl(SparkMakeTime::new()))),
+        // Registry UDFs (including datafusion-spark) cannot receive fail_on_error.
+        _ if fail_on_error => Err(DataFusionError::Execution(format!(
+            "Function '{fun_name}' is resolved from the UDF registry and cannot \
+             honor fail_on_error=true. Use a name-based ANSI/try variant \
+             (e.g. parse_url / try_parse_url) or a dedicated match arm that \
+             consumes the flag."
+        ))),
         _ => registry.udf(fun_name).map_err(|e| {
             DataFusionError::Execution(format!(
                 "Function {fun_name} not found in the registry: {e}",

native/spark-expr/tests/spark_expr_reg.rs

@@ -17,7 +17,7 @@
 
 #[cfg(test)]
 mod tests {
-    use arrow::datatypes::DataType;
+    use arrow::datatypes::{DataType, TimeUnit};
     use datafusion::error::Result;
     use datafusion::execution::session_state::SessionStateBuilder;
     use datafusion::execution::FunctionRegistry;
@@ -91,4 +91,68 @@ mod tests {
 
         Ok(())
     }
+
+    /// Dedicated match arm must honor fail_on_error=true at execution time (#5074).
+    #[tokio::test]
+    async fn test_make_date_ansi_invalid_input_errors() -> Result<()> {
+        let ctx = SessionContext::new();
+        let udf = {
+            let session_state = ctx.state();
+            create_comet_physical_fun("make_date", DataType::Date32, &session_state, Some(true))?
+        };
+        ctx.register_udf(udf.as_ref().clone());
+
+        let df = ctx.sql("SELECT make_date(2023, 0, 15)").await?;
+        let result = df.collect().await;
+        assert!(
+            result.is_err(),
+            "make_date with fail_on_error=true must error on invalid date, got Ok"
+        );
+
+        Ok(())
+    }
+
+    /// Registry catch-all must fail closed when fail_on_error=true (#5074).
+    #[tokio::test]
+    async fn test_registry_udf_rejects_fail_on_error() -> Result<()> {
+        // SessionContext registers DataFusion built-ins (e.g. acos) used by the catch-all.
+        let ctx = SessionContext::new();
+        let session_state = ctx.state();
+        // "acos" is a DataFusion built-in resolved only via the registry catch-all.
+        let err = create_comet_physical_fun("acos", DataType::Float64, &session_state, Some(true));
+        assert!(
+            err.is_err(),
+            "registry UDF must reject fail_on_error=true, got Ok"
+        );
+        let msg = err.unwrap_err().to_string();
+        assert!(msg.contains("acos"), "unexpected error message: {msg}");
+        assert!(
+            msg.contains("fail_on_error=true"),
+            "unexpected error message: {msg}"
+        );
+
+        let udf_false =
+            create_comet_physical_fun("acos", DataType::Float64, &session_state, Some(false))?;
+        assert_eq!(udf_false.name(), "acos");
+
+        let udf_none = create_comet_physical_fun("acos", DataType::Float64, &session_state, None)?;
+        assert_eq!(udf_none.name(), "acos");
+
+        // Dedicated match arms that consume the flag still accept fail_on_error=true.
+        let make_date =
+            create_comet_physical_fun("make_date", DataType::Date32, &session_state, Some(true))?;
+        assert_eq!(make_date.name(), "make_date");
+
+        // Spark 4.1+ make_time serde always passes fail_on_error=true; the dedicated arm
+        // must accept it even though SparkMakeTime does not take the flag as a constructor arg.
+        let make_time = create_comet_physical_fun(
+            "make_time",
+            DataType::Time64(TimeUnit::Nanosecond),
+            &session_state,
+            Some(true),
+        )?;
+        assert_eq!(make_time.name(), "make_time");
+
+        Ok(())
+    }
 }

spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala

@@ -21,14 +21,60 @@ package org.apache.comet.serde
 
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
 
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto}
 
 /** Serde for scalar function. */
 case class CometScalarFunction[T <: Expression](name: String) extends CometExpressionSerde[T] {
   override def convert(expr: T, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = {
+    if (CometScalarFunction.isAnsiSensitive(expr)) {
+      withFallbackReason(
+        expr,
+        s"${expr.nodeName} carries failOnError/evalMode/nullOnOverflow/ansiEnabled/evalContext " +
+          s"and cannot use CometScalarFunction('$name'). Prefer name-based ANSI/try variants " +
+          "(e.g. parse_url / try_parse_url), or a custom serde with " +
+          "scalarFunctionExprToProtoWithReturnType plus a native match arm that " +
+          "consumes fail_on_error.")
+      return None
+    }
     val childExpr = expr.children.map(exprToProtoInternal(_, inputs, binding))
     val optExpr = scalarFunctionExprToProto(name, childExpr: _*)
     optExpr
   }
 }
+
+object CometScalarFunction {
+
+  /**
+   * Product / Java field names that indicate ANSI / eval-mode sensitive Spark expressions.
+   *
+   * Detection uses Java reflection (not `Product.productElementNames`) so the check compiles on
+   * Scala 2.12 used by Spark 3.4 / 3.5.
+   */
+  private val AnsiSensitiveFields: Set[String] =
+    Set("failOnError", "evalMode", "nullOnOverflow", "ansiEnabled", "evalContext")
+
+  /**
+   * True when the Spark expression case class declares an ANSI-related constructor field. Used to
+   * reject miswiring via plain [[CometScalarFunction]].
+   */
+  private[serde] def isAnsiSensitive(expr: Expression): Boolean = {
+    isAnsiSensitive(expr.getClass)
+  }
+
+  /**
+   * Class-level check used by registration audits: the Spark expression type carries an
+   * ANSI-related field regardless of any particular instance's flag value.
+   */
+  private[serde] def isAnsiSensitive(clazz: Class[_]): Boolean = {
+    var current: Class[_] = clazz
+    while (current != null && current != classOf[Object]) {
+      if (current.getDeclaredFields.exists(f => AnsiSensitiveFields.contains(f.getName))) {
+        return true
+      }
+      current = current.getSuperclass
+    }
+    false
+  }
+}