diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index b42d80d3482..d42a4668067 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.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 diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 71a7e94c5b8..ed63ba413e0 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.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 }}] diff --git a/docs/source/contributor-guide/adding_a_new_expression.md b/docs/source/contributor-guide/adding_a_new_expression.md index 66d2e8b3b18..2a92139830d 100644 --- a/docs/source/contributor-guide/adding_a_new_expression.md +++ b/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. diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs index e90e01de953..dcb6b1906ce 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/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}", diff --git a/native/spark-expr/tests/spark_expr_reg.rs b/native/spark-expr/tests/spark_expr_reg.rs index 633b226068f..1049f409b00 100644 --- a/native/spark-expr/tests/spark_expr_reg.rs +++ b/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(()) + } } diff --git a/spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala b/spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala index e5f14f8b015..87db1da040d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala +++ b/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 + } +} diff --git a/spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala b/spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala new file mode 100644 index 00000000000..8cf88ec5593 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/serde/CometScalarFunctionSuite.scala @@ -0,0 +1,245 @@ +/* + * 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 org.apache.comet.serde + +import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.catalyst.expressions.{Abs, Cos, Expression, Literal, Round, Unevaluable} +import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType} + +import org.apache.comet.{CometExplainInfo, CometSparkSessionExtensions} + +/** + * Synthetic expression whose constructor declares `evalMode`, used to prove class-level detection + * without depending on Spark-version-specific arithmetic field names. + */ +case class TestEvalModeExpression(child: Expression, evalMode: Boolean) + extends Expression + with Unevaluable { + override def children: Seq[Expression] = Seq(child) + override def nullable: Boolean = child.nullable + override def dataType: DataType = IntegerType + override protected def withNewChildrenInternal( + newChildren: IndexedSeq[Expression]): Expression = + copy(child = newChildren.head) +} + +/** + * Synthetic expression whose constructor declares `nullOnOverflow`, matching markers such as + * Spark's `MakeDecimal`. + */ +case class TestNullOnOverflowExpression(child: Expression, nullOnOverflow: Boolean) + extends Expression + with Unevaluable { + override def children: Seq[Expression] = Seq(child) + override def nullable: Boolean = child.nullable + override def dataType: DataType = IntegerType + override protected def withNewChildrenInternal( + newChildren: IndexedSeq[Expression]): Expression = + copy(child = newChildren.head) +} + +/** + * Synthetic expression whose constructor declares `ansiEnabled`, matching markers such as Spark's + * `Round` / `BRound` / `Conv`. + */ +case class TestAnsiEnabledExpression(child: Expression, ansiEnabled: Boolean) + extends Expression + with Unevaluable { + override def children: Seq[Expression] = Seq(child) + override def nullable: Boolean = child.nullable + override def dataType: DataType = IntegerType + override protected def withNewChildrenInternal( + newChildren: IndexedSeq[Expression]): Expression = + copy(child = newChildren.head) +} + +/** + * Synthetic expression whose constructor declares `evalContext`, matching Spark 4.1+ arithmetic + * expressions that store `NumericEvalContext` as a field. + */ +case class TestEvalContextExpression(child: Expression, evalContext: Any) + extends Expression + with Unevaluable { + override def children: Seq[Expression] = Seq(child) + override def nullable: Boolean = child.nullable + override def dataType: DataType = IntegerType + override protected def withNewChildrenInternal( + newChildren: IndexedSeq[Expression]): Expression = + copy(child = newChildren.head) +} + +class CometScalarFunctionSuite extends CometTestBase { + + private def fallbackReasons(expr: Expression): Set[String] = { + expr.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty) + } + + private def assertRejectReason(expr: Expression, expectedTokens: String*): Unit = { + val reasons = fallbackReasons(expr) + assert(reasons.nonEmpty, s"expected fallback reason on ${expr.nodeName}") + val joined = reasons.mkString(" ") + expectedTokens.foreach { token => + assert(joined.contains(token), s"expected '$token' in fallback reasons: $reasons") + } + } + + test("CometScalarFunction rejects ANSI-sensitive expressions (#5074)") { + val abs = Abs(Literal(1), failOnError = true) + val result = CometScalarFunction[Abs]("abs").convert(abs, Seq.empty, binding = true) + assert(result.isEmpty) + assertRejectReason(abs, abs.nodeName, "CometScalarFunction", "failOnError") + } + + test("rejects expressions with failOnError even when false (#5074)") { + val abs = Abs(Literal(1), failOnError = false) + val result = CometScalarFunction[Abs]("abs").convert(abs, Seq.empty, binding = true) + assert(result.isEmpty) + assertRejectReason(abs, abs.nodeName, "CometScalarFunction", "failOnError") + } + + test("isAnsiSensitive detects failOnError field") { + assert(CometScalarFunction.isAnsiSensitive(Abs(Literal(1), failOnError = false))) + assert(CometScalarFunction.isAnsiSensitive(classOf[Abs])) + assert(!CometScalarFunction.isAnsiSensitive(Cos(Literal(0.0)))) + assert(!CometScalarFunction.isAnsiSensitive(classOf[Cos])) + } + + test("isAnsiSensitive detects evalMode field") { + val ansi = TestEvalModeExpression(Literal(1), evalMode = true) + val legacy = TestEvalModeExpression(Literal(1), evalMode = false) + assert(CometScalarFunction.isAnsiSensitive(ansi)) + assert(CometScalarFunction.isAnsiSensitive(legacy)) + assert(CometScalarFunction.isAnsiSensitive(classOf[TestEvalModeExpression])) + } + + test("isAnsiSensitive detects nullOnOverflow field") { + val nullOnOverflow = TestNullOnOverflowExpression(Literal(1), nullOnOverflow = true) + val failOnOverflow = TestNullOnOverflowExpression(Literal(1), nullOnOverflow = false) + assert(CometScalarFunction.isAnsiSensitive(nullOnOverflow)) + assert(CometScalarFunction.isAnsiSensitive(failOnOverflow)) + assert(CometScalarFunction.isAnsiSensitive(classOf[TestNullOnOverflowExpression])) + } + + test("isAnsiSensitive detects ansiEnabled field") { + val enabled = TestAnsiEnabledExpression(Literal(1), ansiEnabled = true) + val disabled = TestAnsiEnabledExpression(Literal(1), ansiEnabled = false) + assert(CometScalarFunction.isAnsiSensitive(enabled)) + assert(CometScalarFunction.isAnsiSensitive(disabled)) + assert(CometScalarFunction.isAnsiSensitive(classOf[TestAnsiEnabledExpression])) + } + + test("isAnsiSensitive detects evalContext field") { + val withContext = TestEvalContextExpression(Literal(1), evalContext = "legacy") + assert(CometScalarFunction.isAnsiSensitive(withContext)) + assert(CometScalarFunction.isAnsiSensitive(classOf[TestEvalContextExpression])) + } + + test("isAnsiSensitive detects Spark Round ansiEnabled") { + val round = Round(Literal(1.5, DoubleType), Literal(0)) + assert(CometScalarFunction.isAnsiSensitive(round)) + assert(CometScalarFunction.isAnsiSensitive(classOf[Round])) + assert( + CometScalarFunction[Round]("round") + .convert(round, Seq.empty, binding = true) + .isEmpty) + assertRejectReason(round, "CometScalarFunction", "ansiEnabled") + } + + test("rejects expressions with evalMode via plain CometScalarFunction (#5074)") { + val ansi = TestEvalModeExpression(Literal(1), evalMode = true) + val legacy = TestEvalModeExpression(Literal(1), evalMode = false) + assert( + CometScalarFunction[TestEvalModeExpression]("test") + .convert(ansi, Seq.empty, binding = true) + .isEmpty) + assert( + CometScalarFunction[TestEvalModeExpression]("test") + .convert(legacy, Seq.empty, binding = true) + .isEmpty) + assertRejectReason(ansi, "CometScalarFunction", "evalMode") + assertRejectReason(legacy, "CometScalarFunction", "evalMode") + } + + test("rejects expressions with nullOnOverflow via plain CometScalarFunction (#5074)") { + val nullOnOverflow = TestNullOnOverflowExpression(Literal(1), nullOnOverflow = true) + val failOnOverflow = TestNullOnOverflowExpression(Literal(1), nullOnOverflow = false) + assert( + CometScalarFunction[TestNullOnOverflowExpression]("test") + .convert(nullOnOverflow, Seq.empty, binding = true) + .isEmpty) + assert( + CometScalarFunction[TestNullOnOverflowExpression]("test") + .convert(failOnOverflow, Seq.empty, binding = true) + .isEmpty) + assertRejectReason(nullOnOverflow, "CometScalarFunction", "nullOnOverflow") + assertRejectReason(failOnOverflow, "CometScalarFunction", "nullOnOverflow") + } + + test("rejects expressions with ansiEnabled via plain CometScalarFunction") { + val enabled = TestAnsiEnabledExpression(Literal(1), ansiEnabled = true) + val disabled = TestAnsiEnabledExpression(Literal(1), ansiEnabled = false) + assert( + CometScalarFunction[TestAnsiEnabledExpression]("test") + .convert(enabled, Seq.empty, binding = true) + .isEmpty) + assert( + CometScalarFunction[TestAnsiEnabledExpression]("test") + .convert(disabled, Seq.empty, binding = true) + .isEmpty) + assertRejectReason(enabled, "CometScalarFunction", "ansiEnabled") + assertRejectReason(disabled, "CometScalarFunction", "ansiEnabled") + } + + test("rejects expressions with evalContext via plain CometScalarFunction") { + val withContext = TestEvalContextExpression(Literal(1), evalContext = "ansi") + assert( + CometScalarFunction[TestEvalContextExpression]("test") + .convert(withContext, Seq.empty, binding = true) + .isEmpty) + assertRejectReason(withContext, "CometScalarFunction", "evalContext") + } + + test("CometScalarFunction allows non-ANSI expressions") { + val cos = Cos(Literal(0.0)) + val result = CometScalarFunction[Cos]("cos").convert(cos, Seq.empty, binding = true) + assert(result.isDefined) + val proto = result.get + assert(proto.hasScalarFunc) + assert(proto.getScalarFunc.getFunc === "cos") + assert(!proto.getScalarFunc.getFailOnError) + assert(proto.getScalarFunc.getArgsCount === 1) + assert(!CometSparkSessionExtensions.hasFallbackReason(cos)) + } + + test("no ANSI-sensitive expression is registered with plain CometScalarFunction") { + val violations = QueryPlanSerde.exprSerdeMap + .collect { + case (sparkClass, _: CometScalarFunction[_]) + if CometScalarFunction.isAnsiSensitive(sparkClass) => + sparkClass.getName + } + .toSeq + .sorted + assert( + violations.isEmpty, + "ANSI-sensitive expressions use plain CometScalarFunction: " + + violations.mkString(", ")) + } +} diff --git a/spark/src/test/spark-4.1+/org/apache/spark/sql/comet/CometDecimalArithmeticViewSuite.scala b/spark/src/test/spark-4.1+/org/apache/spark/sql/comet/CometDecimalArithmeticViewSuite.scala index 0a214cf91f8..3e68ffb80b0 100644 --- a/spark/src/test/spark-4.1+/org/apache/spark/sql/comet/CometDecimalArithmeticViewSuite.scala +++ b/spark/src/test/spark-4.1+/org/apache/spark/sql/comet/CometDecimalArithmeticViewSuite.scala @@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference, Binar import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.DecimalType -import org.apache.comet.serde.{ExprOuterClass, QueryPlanSerde} +import org.apache.comet.serde.{CometScalarFunction, ExprOuterClass, QueryPlanSerde} class CometDecimalArithmeticViewSuite extends CometTestBase { @@ -104,4 +104,15 @@ class CometDecimalArithmeticViewSuite extends CometTestBase { } } } + + test("plain CometScalarFunction rejects Spark 4.1+ Add with evalContext") { + val left = AttributeReference("a", DecimalType(10, 0))() + val right = AttributeReference("b", DecimalType(10, 0))() + val add = + Add(left, right, NumericEvalContext(EvalMode.LEGACY, allowDecimalPrecisionLoss = true)) + assert( + CometScalarFunction[Add]("add") + .convert(add, Seq(left, right), binding = true) + .isEmpty) + } }