From 62c841993f515921723cbde5d3fc371c558efc27 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 3 Aug 2026 13:34:10 -0600 Subject: [PATCH 1/4] refactor: replace hand-coded fallback-reason roll-up with a central traversal Comet records fallback reasons in a `TreeNodeTag` side channel. Extended explain output only walks plan nodes, so an expression-level reason is invisible unless something lifts it onto the enclosing operator. That lifting was hand-written at ~200 call sites, and nothing forced it, so forgetting the roll-up argument silently produced a plausible-looking generic message instead of the real reason (fixed twice before, in #2323 and #2716). Steps 1 and 2 of #5230: Strict mode. `CometExecRule.reportUnexplainedFallback` replaces the masking behaviour: when Comet declines an operator whose children are all already native, and neither the operator nor any of its expressions carries a reason, `spark.comet.explain.fallback.strict.enabled` (internal, default off) throws instead of tagging `" is not supported"`. `CometTestBase` enables it, so the whole test corpus now enforces it. The check is scoped to operators a serde actually attempted; an operator with no registered handler was never attempted and still gets the generic message. Central traversal. `CometExecRule.rollUpFallbackReasons` collects `FALLBACK_REASONS` from `op.expressions` and tags them on the operator at the single point where Comet decides to keep the Spark operator, mirroring the existing `rollUpInfoMessages`. It only runs on the operator that failed conversion, which contains the shared-instance problem for `AttributeReference`s and DPP subquery expressions. `hasFallbackReason` still reads only the node's own tag: it is a planning control signal and must not observe the traversal. With that in place the roll-up parameters are dead, so they are gone from the API: `withFallbackReason(node, info)` and `withFallbackReasons(node, info)` no longer take varargs, and the pure-roll-up overload and `optExprWithFallbackReason` are deleted. The compiler now rejects any attempt to reintroduce a hand-rolled roll-up, and the old signature's lack of type safety (issue point 4) goes with it. Also drops the `var allProjExprs` accumulator in `CometExpandExec`, which existed only to feed the roll-up. Tested: CometExecSuite, CometExpressionSuite, CometAggregateSuite, CometJoinSuite, CometWindowExecSuite, CometGenerateExecSuite, CometExecRuleSuite, CometScanRuleSuite, CometSparkSessionExtensionsSuite, CometFuzzTestSuite, CometFuzzAggregateSuite, CometCastSuite, CometArrayExpressionSuite, CometStringExpressionSuite, CometShuffleSuite, CometNativeShuffleSuite, CometShuffleFallbackStickinessSuite, CometDppFallbackRepro3949Suite, CometCodegenSuite - all pass with strict mode on. Compiles clean on spark-3.4, 3.5, 4.0 and 4.1. --- .../scala/org/apache/comet/CometConf.scala | 14 ++++ .../comet/CometSparkSessionExtensions.scala | 62 ++++++---------- .../apache/comet/expressions/CometCast.scala | 1 - .../apache/comet/rules/CometExecRule.scala | 71 ++++++++++++++++++- .../serde/CometBloomFilterMightContain.scala | 2 - .../comet/serde/CometScalarFunction.scala | 4 +- .../apache/comet/serde/CometSortOrder.scala | 2 - .../apache/comet/serde/QueryPlanSerde.scala | 25 +------ .../org/apache/comet/serde/aggregates.scala | 50 ++++--------- .../org/apache/comet/serde/arithmetic.scala | 8 +-- .../scala/org/apache/comet/serde/arrays.scala | 57 ++++++--------- .../org/apache/comet/serde/bitwise.scala | 4 +- .../org/apache/comet/serde/conditional.scala | 6 -- .../comet/serde/contraintExpressions.scala | 8 +-- .../org/apache/comet/serde/datetime.scala | 33 ++++----- .../comet/serde/decimalExpressions.scala | 6 +- .../scala/org/apache/comet/serde/json.scala | 4 +- .../scala/org/apache/comet/serde/maps.scala | 10 +-- .../scala/org/apache/comet/serde/math.scala | 26 ++++--- .../apache/comet/serde/namedExpressions.scala | 4 +- .../org/apache/comet/serde/predicates.scala | 5 +- .../org/apache/comet/serde/statics.scala | 11 ++- .../org/apache/comet/serde/strings.scala | 20 +++--- .../org/apache/comet/serde/structs.scala | 5 +- .../org/apache/comet/serde/unixtime.scala | 5 +- .../scala/org/apache/comet/serde/url.scala | 4 +- .../spark/sql/comet/CometWindowExec.scala | 32 +++------ .../apache/spark/sql/comet/operators.scala | 34 ++------- .../comet/serde/CometToPrettyString.scala | 2 - .../apache/comet/shims/CometExprShim.scala | 8 +-- .../org/apache/comet/serde/CometMapSort.scala | 4 +- .../comet/serde/CometToPrettyString.scala | 2 - .../apache/comet/shims/CometExprShim4x.scala | 8 +-- .../apache/comet/CometExpressionSuite.scala | 14 ++-- .../comet/rules/CometExecRuleSuite.scala | 60 +++++++++++++++- .../org/apache/spark/sql/CometTestBase.scala | 4 ++ 36 files changed, 307 insertions(+), 308 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index cdbc430829..b3bc2d3312 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -668,6 +668,20 @@ object CometConf extends ShimCometConf { .booleanConf .createWithDefault(false) + val COMET_STRICT_FALLBACK_REASONS: ConfigEntry[Boolean] = + conf("spark.comet.explain.fallback.strict.enabled") + .category(CATEGORY_TESTING) + .doc( + "Test-only. When enabled, Comet throws if it declines to convert an operator that it " + + "could otherwise have converted (all children are already native) without recording a " + + "fallback reason on the operator or on any of its expressions. Without this check, a " + + "serde that returns `None` and forgets to state a reason silently produces a generic " + + "' is not supported' message instead of a visible failure. Enabled for all " + + "Comet test suites via `CometTestBase`.") + .internal() + .booleanConf + .createWithDefault(false) + val COMET_ONHEAP_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.exec.onHeap.enabled") .category(CATEGORY_TESTING) diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala index 3511c7e494..1b7f3942e6 100644 --- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala +++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala @@ -274,41 +274,41 @@ object CometSparkSessionExtensions extends Logging { * Record a fallback reason on a `TreeNode` (a Spark operator or expression) explaining why * Comet cannot accelerate it. Reasons recorded here are surfaced in extended explain output * (see `ExtendedExplainInfo`) and, when `COMET_EXPLAIN_FALLBACK_LOG_ENABLED` is enabled, logged - * as warnings. The reasons are also rolled up from child nodes so that the operator that - * remains in the Spark plan carries the reasons from its converted-away subtree. + * as warnings. * * Call this in any code path where Comet decides not to convert a given node - serde `convert` * methods returning `None`, unsupported data types, disabled configs, etc. Do not use this for * informational messages that are not fallback reasons: anything tagged here is treated by the * rules as a signal that the node falls back to Spark. * + * Tag only the node that actually failed, and state a real reason. There is deliberately no way + * to copy reasons from child nodes onto a parent: extended explain only walks plan nodes, so an + * expression-level reason is lifted onto the enclosing operator centrally by + * `CometExecRule.rollUpFallbackReasons` when that operator is left in the Spark plan. See + * https://github.com/apache/datafusion-comet/issues/5230. + * * @param node * The Spark operator or expression that is falling back to Spark. * @param info - * The fallback reason. Optional, may be null or empty - pass empty only when the call is used - * purely to roll up reasons from `exprs`. - * @param exprs - * Child nodes whose own fallback reasons should be rolled up into `node`. Pass the - * sub-expressions or child operators whose failure caused `node` to fall back. + * The fallback reason. Newline-delimited to record more than one reason. * @tparam T * The type of the TreeNode. Typically `SparkPlan`, `AggregateExpression`, or `Expression`. * @return - * `node` with fallback reasons attached (as a side effect on its tag map). + * `node` with the fallback reason attached (as a side effect on its tag map). */ - def withFallbackReason[T <: TreeNode[_]](node: T, info: String, exprs: T*): T = { + def withFallbackReason[T <: TreeNode[_]](node: T, info: String): T = { // support existing approach of passing in multiple infos in a newline-delimited string val infoSet = if (info == null || info.isEmpty) { Set.empty[String] } else { info.split("\n").toSet } - withFallbackReasons(node, infoSet, exprs: _*) + withFallbackReasons(node, infoSet) } /** - * Record one or more fallback reasons on a `TreeNode` and roll up reasons from any child nodes. - * This is the set-valued form of [[withFallbackReason]]; see that overload for the full - * contract. + * Record one or more fallback reasons on a `TreeNode`. This is the set-valued form of + * [[withFallbackReason]]; see that overload for the full contract. * * Reasons are accumulated (never overwritten) on the node's `FALLBACK_REASONS` tag and are * surfaced in extended explain output. When `COMET_EXPLAIN_FALLBACK_LOG_ENABLED` is enabled, @@ -317,50 +317,32 @@ object CometSparkSessionExtensions extends Logging { * @param node * The Spark operator or expression that is falling back to Spark. * @param info - * The fallback reasons for this node. May be empty when the call is used purely to roll up - * child reasons. - * @param exprs - * Child nodes whose own fallback reasons should be rolled up into `node`. + * The fallback reasons for this node. * @tparam T * The type of the TreeNode. Typically `SparkPlan`, `AggregateExpression`, or `Expression`. * @return * `node` with fallback reasons attached (as a side effect on its tag map). */ - def withFallbackReasons[T <: TreeNode[_]](node: T, info: Set[String], exprs: T*): T = { + def withFallbackReasons[T <: TreeNode[_]](node: T, info: Set[String]): T = { if (CometConf.COMET_EXPLAIN_FALLBACK_LOG_ENABLED.get()) { for (reason <- info) { logWarning(s"Comet cannot accelerate ${node.getClass.getSimpleName} because: $reason") } } - val existingNodeInfos = node.getTagValue(CometExplainInfo.FALLBACK_REASONS) - val newNodeInfo = (existingNodeInfos ++ exprs - .flatMap(_.getTagValue(CometExplainInfo.FALLBACK_REASONS))).flatten.toSet - node.setTagValue(CometExplainInfo.FALLBACK_REASONS, newNodeInfo ++ info) + val existingNodeInfos = + node.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String]) + node.setTagValue(CometExplainInfo.FALLBACK_REASONS, existingNodeInfos ++ info) node } - /** - * Roll up fallback reasons from `exprs` onto `node` without adding a new reason of its own. Use - * this when a parent operator is itself falling back and wants to preserve the reasons recorded - * on its child expressions/operators so they appear together in explain output. - * - * @param node - * The parent operator or expression falling back to Spark. - * @param exprs - * Child nodes whose fallback reasons should be aggregated onto `node`. - * @tparam T - * The type of the TreeNode. Typically `SparkPlan`, `AggregateExpression`, or `Expression`. - * @return - * `node` with the rolled-up reasons attached (as a side effect on its tag map). - */ - def withFallbackReason[T <: TreeNode[_]](node: T, exprs: T*): T = { - withFallbackReasons(node, Set.empty, exprs: _*) - } - /** * True if any fallback reason has been recorded on `node` (via [[withFallbackReason]] / * [[withFallbackReasons]]). Callers that need to short-circuit when a prior rule pass has * already decided a node falls back can use this as the sticky signal. + * + * This deliberately reads only the node's own tag. It is a planning control signal, not explain + * output, so it must not observe reasons that merely exist somewhere in the node's expression + * trees - see `CometExecRule.rollUpFallbackReasons`. */ def hasFallbackReason(node: TreeNode[_]): Boolean = { node.getTagValue(CometExplainInfo.FALLBACK_REASONS).exists(_.nonEmpty) diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala index 78abf54c8d..df4140078e 100644 --- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala +++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala @@ -118,7 +118,6 @@ object CometCast if (childExpr.isDefined) { castToProto(cast, cast.timeZoneId, cast.dataType, childExpr.get, cometEvalMode) } else { - withFallbackReason(cast, cast.child) None } } diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala index f7913e3a4f..1d1b3cf9a4 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala @@ -382,7 +382,11 @@ case class CometExecRule(session: SparkSession) // when COMET_EXPLAIN_FALLBACK_LOG_ENABLED=true) even when the write is fully native. op case _ => - // The operator was not converted to a Comet plan. Possible reasons for this happening: + // The operator was not converted to a Comet plan and no serde handler claimed it, so + // Comet simply has no support for it. (Operators that do have a handler are reported + // by `reportUnexplainedFallback` inside `convertToComet`, which is also where the + // strict check lives - it would be wrong to demand a specific reason here, because + // nothing ever attempted this operator.) Possible reasons for reaching this point: // 1. Comet does not support this operator. // 2. The operator could not be supported based on query context and current // configs. In this case, it should have already been tagged with fallback @@ -698,6 +702,20 @@ case class CometExecRule(session: SparkSession) /** Convert a Spark plan to a Comet plan using the specified serde handler */ private def convertToComet(op: SparkPlan, handler: CometOperatorSerde[_]): Option[SparkPlan] = { + val converted = tryConvertToComet(op, handler) + if (converted.isEmpty) { + // Comet looked at this operator and declined it, so it stays in the Spark plan. Lift any + // reasons recorded on its expressions onto the operator itself - see + // `rollUpFallbackReasons` for why this is needed - and make sure something was recorded. + rollUpFallbackReasons(op) + reportUnexplainedFallback(op) + } + converted + } + + private def tryConvertToComet( + op: SparkPlan, + handler: CometOperatorSerde[_]): Option[SparkPlan] = { val serde = handler.asInstanceOf[CometOperatorSerde[SparkPlan]] if (isOperatorEnabled(serde, op)) { // For operators that require native children (like writes), check if all data-producing @@ -741,6 +759,57 @@ case class CometExecRule(session: SparkSession) None } + /** + * Lift fallback reasons recorded on `op`'s expression trees onto `op` itself. + * + * Extended explain output only walks plan nodes (`ExtendedExplainInfo.sortup` follows + * `children` / `innerChildren`, never `expressions`), so a reason tagged on an expression is + * invisible unless something lifts it onto the enclosing operator. This mirrors what + * [[rollUpInfoMessages]] already does for the informational tags, and replaces the roll-up that + * used to be hand-written at every serde call site (see + * https://github.com/apache/datafusion-comet/issues/5230). + * + * Only child *expressions* are collected, not child operators: reasons on a child operator are + * already reachable by the explain traversal via `children`. + * + * Called only when `op` was left in the Spark plan, which scopes the roll-up to the operator + * that actually failed conversion. That matters because some expression instances + * (`AttributeReference`s, DPP subquery expressions) are shared across operators, so an unscoped + * roll-up could surface one expression's reason under several unrelated operators. + */ + private def rollUpFallbackReasons(op: SparkPlan): Unit = { + val reasons = op.expressions + .flatMap(_.collect { case e: Expression => e }) + .flatMap(_.getTagValue(CometExplainInfo.FALLBACK_REASONS)) + .flatten + .toSet + if (reasons.nonEmpty) { + withFallbackReasons(op, reasons) + } + } + + /** + * Handle an operator that Comet declined without stating why. + * + * When every child is already native, Comet had a real opportunity to convert `op`, so the + * absence of any reason - on `op` or anywhere in its expression trees - means a serde returned + * `None` and forgot to record one. Under `COMET_STRICT_FALLBACK_REASONS` (enabled for Comet's + * own test suites) that is a hard failure; otherwise fall back to a generic message so users + * still see something. The generic message is what used to mask this whole class of bug, which + * is why the strict check exists. + */ + private def reportUnexplainedFallback(op: SparkPlan): Unit = { + if (op.children.forall(_.isInstanceOf[CometNativeExec]) && !hasFallbackReason(op)) { + if (CometConf.COMET_STRICT_FALLBACK_REASONS.get(op.conf)) { + throw new IllegalStateException( + s"Comet did not convert ${op.nodeName} but recorded no fallback reason on the " + + "operator or any of its expressions. Add a withFallbackReason call stating why " + + s"conversion failed. Operator:\n$op") + } + withFallbackReason(op, s"${op.nodeName} is not supported") + } + } + /** * Lift informational (non-fallback) messages tagged on an operator and its expressions onto the * converted Comet plan node so they appear in verbose extended explain output. Expression-level diff --git a/spark/src/main/scala/org/apache/comet/serde/CometBloomFilterMightContain.scala b/spark/src/main/scala/org/apache/comet/serde/CometBloomFilterMightContain.scala index 75e0f532e3..9620f2d8fc 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometBloomFilterMightContain.scala +++ b/spark/src/main/scala/org/apache/comet/serde/CometBloomFilterMightContain.scala @@ -21,7 +21,6 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, BloomFilterMightContain} -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal object CometBloomFilterMightContain extends CometExpressionSerde[BloomFilterMightContain] { @@ -45,7 +44,6 @@ object CometBloomFilterMightContain extends CometExpressionSerde[BloomFilterMigh .setBloomFilterMightContain(builder) .build()) } else { - withFallbackReason(expr, bloomFilter, value) None } } 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 2e056ae886..e5f14f8b01 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala +++ b/spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala @@ -22,13 +22,13 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} import org.apache.comet.serde.ExprOuterClass.Expr -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} +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] = { val childExpr = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProto(name, childExpr: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala b/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala index 346bb454eb..838a229bb9 100644 --- a/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala +++ b/spark/src/main/scala/org/apache/comet/serde/CometSortOrder.scala @@ -22,7 +22,6 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, Descending, NullsFirst, NullsLast, SortOrder} import org.apache.comet.CometConf -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal object CometSortOrder extends CometExpressionSerde[SortOrder] { @@ -66,7 +65,6 @@ object CometSortOrder extends CometExpressionSerde[SortOrder] { .setSortOrder(sortOrderBuilder) .build()) } else { - withFallbackReason(expr, expr.child) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 4ae8144648..87e69084f1 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -723,10 +723,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { aggHandler.convert(aggExpr, fn, inputs, binding, conf) } case _ => - withFallbackReason( - aggExpr, - s"unsupported Spark aggregate function: ${fn.prettyName}", - fn.children: _*) + withFallbackReason(aggExpr, s"unsupported Spark aggregate function: ${fn.prettyName}") None } @@ -740,7 +737,6 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { if (aggExpr.filter.isDefined && aggExpr.mode == Partial) { val filterProto = exprToProto(aggExpr.filter.get, inputs, binding) if (filterProto.isEmpty) { - withFallbackReason(aggExpr, aggExpr.filter.get) return None } builder.setFilter(filterProto.get) @@ -893,7 +889,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { case Some(handler) => convert(expr, handler.asInstanceOf[CometExpressionSerde[Expression]]) case _ => - withFallbackReason(expr, s"${expr.prettyName} is not supported", expr.children: _*) + withFallbackReason(expr, s"${expr.prettyName} is not supported") None } }) @@ -944,7 +940,6 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { .newBuilder(), inner).build()) } else { - withFallbackReason(expr, child) None } } @@ -974,7 +969,6 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { .newBuilder(), inner).build()) } else { - withFallbackReason(expr, left, right) None } } @@ -1003,7 +997,6 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { : Option[ExprOuterClass.Expr] = { val protos = operands.map(exprToProtoInternal(_, inputs, binding)) if (protos.exists(_.isEmpty)) { - withFallbackReason(expr, operands: _*) None } else { val leaves = protos.map(_.get).toIndexedSeq @@ -1086,20 +1079,6 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { Some(ExprOuterClass.Expr.newBuilder().setScalarFunc(builder).build()) } - // Utility method. Adds fallback reason if the result of calling exprToProto is None - def optExprWithFallbackReason( - optExpr: Option[Expr], - expr: Expression, - childExpr: Expression*): Option[Expr] = { - optExpr match { - case None => - withFallbackReason(expr, childExpr: _*) - None - case o => o - } - - } - /** * If `handler` is a `CodegenDispatchFallback`, run `expr` through the JVM codegen dispatcher * and return `Some((handler, proto))` on success; otherwise return `None`. Shared by the diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index 316744be1f..db435e5b5d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -61,10 +61,9 @@ object CometMin extends CometAggregateExpressionSerde[Min] { .setMin(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -98,10 +97,9 @@ object CometMax extends CometAggregateExpressionSerde[Max] { .setMax(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -124,7 +122,6 @@ object CometCount extends CometAggregateExpressionSerde[Count] { .setCount(builder) .build()) } else { - withFallbackReason(aggExpr, expr.children: _*) None } } @@ -182,10 +179,9 @@ object CometAverage extends CometAggregateExpressionSerde[Average] { .setAvg(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${avg.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${avg.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -232,9 +228,7 @@ object CometSum extends CometAggregateExpressionSerde[Sum] { .build()) } else { if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${sum.dataType} is not supported", sum.child) - } else { - withFallbackReason(aggExpr, sum.child) + withFallbackReason(aggExpr, s"datatype ${sum.dataType} is not supported") } None } @@ -268,10 +262,9 @@ object CometFirst extends CometAggregateExpressionSerde[First] { .setFirst(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${first.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${first.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -304,10 +297,9 @@ object CometLast extends CometAggregateExpressionSerde[Last] { .setLast(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${last.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${last.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -343,10 +335,9 @@ object CometBitAndAgg extends CometAggregateExpressionSerde[BitAndAgg] { .setBitAndAgg(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${bitAnd.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${bitAnd.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -382,10 +373,9 @@ object CometBitOrAgg extends CometAggregateExpressionSerde[BitOrAgg] { .setBitOrAgg(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${bitOr.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${bitOr.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -421,10 +411,9 @@ object CometBitXOrAgg extends CometAggregateExpressionSerde[BitXorAgg] { .setBitXorAgg(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${bitXor.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${bitXor.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -523,7 +512,6 @@ trait CometVariance { .setVariance(builder) .build()) } else { - withFallbackReason(aggExpr, expr.child) None } } @@ -578,7 +566,6 @@ trait CometStddev { .setStddev(builder) .build()) } else { - withFallbackReason(aggExpr, child) None } } @@ -671,7 +658,6 @@ object CometPercentile extends CometAggregateExpressionSerde[Percentile] { .setPercentile(builder) .build()) } else { - withFallbackReason(aggExpr, percentile.child) None } } @@ -719,14 +705,14 @@ object CometApproxPercentile extends CometAggregateExpressionSerde[ApproximatePe case d: Double => (Seq(d), false) case arr: ArrayData => (arr.toDoubleArray().toSeq, true) case other => - withFallbackReason(aggExpr, s"Unsupported percentage literal: $other", expr.child) + withFallbackReason(aggExpr, s"Unsupported percentage literal: $other") return None } val accuracy = expr.accuracyExpression.eval() match { case i: Int => i.toLong case l: Long => l case other => - withFallbackReason(aggExpr, s"Unsupported accuracy literal: $other", expr.child) + withFallbackReason(aggExpr, s"Unsupported accuracy literal: $other") return None } @@ -743,7 +729,6 @@ object CometApproxPercentile extends CometAggregateExpressionSerde[ApproximatePe .setApproxPercentile(builder) .build()) } else { - withFallbackReason(aggExpr, expr.child) None } } @@ -773,7 +758,6 @@ object CometCorr extends CometAggregateExpressionSerde[Corr] { .setCorrelation(builder) .build()) } else { - withFallbackReason(aggExpr, corr.x, corr.y) None } } @@ -839,11 +823,6 @@ object CometBloomFilterAggregate extends CometAggregateExpressionSerde[BloomFilt .setBloomFilterAgg(builder) .build()) } else { - withFallbackReason( - aggExpr, - bloomFilter.child, - bloomFilter.estimatedNumItemsExpression, - bloomFilter.numBitsExpression) None } } @@ -897,10 +876,9 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] { .setCollectSet(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -943,10 +921,9 @@ object CometCollectList extends CometAggregateExpressionSerde[CollectList] { .setCollectList(builder) .build()) } else if (dataType.isEmpty) { - withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported", child) + withFallbackReason(aggExpr, s"datatype ${expr.dataType} is not supported") None } else { - withFallbackReason(aggExpr, child) None } } @@ -1009,7 +986,6 @@ object CometApproxCountDistinct extends CometAggregateExpressionSerde[HyperLogLo .setHllpp(builder) .build()) } else { - withFallbackReason(aggExpr, expr.child) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala b/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala index 614fa4a5c0..67d986d5b5 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arithmetic.scala @@ -24,9 +24,8 @@ import scala.math.min import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, Cast, Divide, EmptyRow, EqualTo, EvalMode, Expression, If, IntegralDivide, Literal, Multiply, Remainder, Round, Subtract, UnaryMinus} import org.apache.spark.sql.types.{ByteType, DataType, DecimalType, DoubleType, FloatType, IntegerType, LongType, ShortType} -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.expressions.{CometCast, CometEvalMode} -import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProtoInternal, flattenAssociative, optExprWithFallbackReason, scalarFunctionExprToProtoWithReturnType, serializeDataType} +import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProtoInternal, flattenAssociative, scalarFunctionExprToProtoWithReturnType, serializeDataType} import org.apache.comet.shims.CometEvalModeUtil trait MathBase { @@ -62,7 +61,6 @@ trait MathBase { .newBuilder(), inner).build()) } else { - withFallbackReason(expr, left, right) None } } @@ -127,7 +125,6 @@ trait MathBase { : Option[ExprOuterClass.Expr] = { val protos = operands.map(exprToProtoInternal(_, inputs, binding)) if (protos.exists(_.isEmpty)) { - withFallbackReason(expr, operands: _*) None } else { val returnType = serializeDataType(dataType) @@ -438,7 +435,7 @@ object CometRound extends CometExpressionSerde[Round] { r.ansiEnabled, childExpr, scaleExpr) - optExprWithFallbackReason(optExpr, r, r.child) + optExpr } } @@ -463,7 +460,6 @@ object CometUnaryMinus extends CometExpressionSerde[UnaryMinus] with MathBase { .setUnaryMinus(builder) .build()) } else { - withFallbackReason(expr, expr.child) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala index dfaecd3786..748b1cee23 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -95,7 +95,6 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] { .setCaseWhen(caseWhenExpr) .build()) } else { - withFallbackReason(expr, expr.children: _*) None } } @@ -171,7 +170,7 @@ object CometSortArray extends CometExpressionSerde[SortArray] with CodegenDispat arrayExprProto, sortDirectionExprProto, nullOrderingExprProto) - optExprWithFallbackReason(sortArrayScalarExpr, expr, expr.children: _*) + sortArrayScalarExpr } } @@ -212,7 +211,7 @@ object CometArrayIntersect val arraysIntersectScalarExpr = scalarFunctionExprToProto("array_intersect", leftArrayExprProto, rightArrayExprProto) - optExprWithFallbackReason(arraysIntersectScalarExpr, expr, expr.children: _*) + arraysIntersectScalarExpr } } @@ -225,7 +224,7 @@ object CometArrayMax extends CometExpressionSerde[ArrayMax] { val arrayMaxScalarExpr = scalarFunctionExprToProto("array_max", arrayExprProto) - optExprWithFallbackReason(arrayMaxScalarExpr, expr) + arrayMaxScalarExpr } } @@ -237,7 +236,7 @@ object CometArrayMin extends CometExpressionSerde[ArrayMin] { val arrayExprProto = exprToProto(expr.children.head, inputs, binding) val arrayMinScalarExpr = scalarFunctionExprToProto("array_min", arrayExprProto) - optExprWithFallbackReason(arrayMinScalarExpr, expr) + arrayMinScalarExpr } } @@ -255,7 +254,7 @@ object CometArraysOverlap extends CometExpressionSerde[ArraysOverlap] { false, leftArrayExprProto, rightArrayExprProto) - optExprWithFallbackReason(arraysOverlapScalarExpr, expr, expr.children: _*) + arraysOverlapScalarExpr } } @@ -269,7 +268,7 @@ object CometArrayCompact extends CometExpressionSerde[Expression] { val arrayExprProto = exprToProto(child, inputs, binding) val arrayCompactScalarExpr = scalarFunctionExprToProto("array_compact", arrayExprProto) - optExprWithFallbackReason(arrayCompactScalarExpr, expr, expr.children: _*) + arrayCompactScalarExpr } } @@ -327,7 +326,7 @@ object CometArrayExcept val arrayExceptScalarExpr = scalarFunctionExprToProto("array_except", leftArrayExprProto, rightArrayExprProto) - optExprWithFallbackReason(arrayExceptScalarExpr, expr, expr.children: _*) + arrayExceptScalarExpr } } @@ -375,17 +374,12 @@ object CometArrayJoin delimiterExprProto, nullReplacementExprProto) - optExprWithFallbackReason( - arrayJoinScalarExpr, - expr, - arrayExpr, - arrayExpr.delimiter, - nullReplacementExpr) + arrayJoinScalarExpr case None => val arrayJoinScalarExpr = scalarFunctionExprToProto("array_to_string", arrayExprProto, delimiterExprProto) - optExprWithFallbackReason(arrayJoinScalarExpr, expr, arrayExpr, arrayExpr.delimiter) + arrayJoinScalarExpr } } } @@ -417,12 +411,7 @@ object CometArrayInsert extends CometExpressionSerde[ArrayInsert] { .setArrayInsert(arrayInsertBuilder) .build()) } else { - withFallbackReason( - expr, - "unsupported arguments for ArrayInsert", - expr.children.head, - expr.children(1), - expr.children(2)) + withFallbackReason(expr, "unsupported arguments for ArrayInsert") None } } @@ -448,7 +437,7 @@ object CometSlice extends CometExpressionSerde[Slice] { arrayExprProto, startExprProto, lengthExprProto) - optExprWithFallbackReason(sliceScalarExpr, expr, expr.children: _*) + sliceScalarExpr } } @@ -462,7 +451,7 @@ object CometArrayUnion extends CometExpressionSerde[ArrayUnion] { val arraysUnionScalarExpr = scalarFunctionExprToProto("array_union", leftArrayExprProto, rightArrayExprProto) - optExprWithFallbackReason(arraysUnionScalarExpr, expr, expr.children: _*) + arraysUnionScalarExpr } } @@ -498,8 +487,7 @@ object CometCreateArray extends CometExpressionSerde[CreateArray] { withFallbackReason( expr, "CreateArray children have mismatched data types: " + - children.map(_.dataType).distinct.mkString(", "), - children: _*) + children.map(_.dataType).distinct.mkString(", ")) return None } @@ -508,7 +496,7 @@ object CometCreateArray extends CometExpressionSerde[CreateArray] { if (childExprs.forall(_.isDefined)) { scalarFunctionExprToProto("make_array", childExprs: _*) } else { - withFallbackReason(expr, "unsupported arguments for CreateArray", children: _*) + withFallbackReason(expr, "unsupported arguments for CreateArray") None } } @@ -557,7 +545,7 @@ object CometGetArrayItem extends CometExpressionSerde[GetArrayItem] { .setListExtract(listExtractBuilder) .build()) } else { - withFallbackReason(expr, "unsupported arguments for GetArrayItem", expr.child, expr.ordinal) + withFallbackReason(expr, "unsupported arguments for GetArrayItem") None } } @@ -595,7 +583,7 @@ object CometArrayReverse extends CometExpressionSerde[Reverse] with ArraysBase { } val reverseExprProto = exprToProto(expr.child, inputs, binding) val reverseScalarExpr = scalarFunctionExprToProto("array_reverse", reverseExprProto) - optExprWithFallbackReason(reverseScalarExpr, expr, expr.children: _*) + reverseScalarExpr } } @@ -620,7 +608,7 @@ object CometElementAt extends CometExpressionSerde[ElementAt] { expr.left.dataType match { case _: MapType => val mapExtractExpr = scalarFunctionExprToProto("map_extract", childExpr, ordinalExpr) - optExprWithFallbackReason(mapExtractExpr, expr, expr.left, expr.right) + mapExtractExpr case _ => val defaultExpr = expr.defaultValueOutOfBound.flatMap(exprToProtoInternal(_, inputs, binding)) @@ -642,7 +630,7 @@ object CometElementAt extends CometExpressionSerde[ElementAt] { .setListExtract(arrayExtractBuilder) .build()) } else { - withFallbackReason(expr, "unsupported arguments for ElementAt", expr.left, expr.right) + withFallbackReason(expr, "unsupported arguments for ElementAt") None } } @@ -659,7 +647,7 @@ object CometFlatten extends CometExpressionSerde[Flatten] with ArraysBase { binding: Boolean): Option[ExprOuterClass.Expr] = { val flattenExprProto = exprToProto(expr.child, inputs, binding) val flattenScalarExpr = scalarFunctionExprToProto("flatten", flattenExprProto) - optExprWithFallbackReason(flattenScalarExpr, expr, expr.children: _*) + flattenScalarExpr } } @@ -762,7 +750,7 @@ object CometArrayPosition extends CometExpressionSerde[ArrayPosition] with Array // (matching Spark's behavior) val optExpr = scalarFunctionExprToProto("spark_array_position", arrayExprProto, elementExprProto) - optExprWithFallbackReason(optExpr, expr, expr.left, expr.right) + optExpr } } @@ -829,10 +817,7 @@ object CometArraysZip extends CometExpressionSerde[ArraysZip] { .build()) } else { - withFallbackReason( - expr, - "unsupported arguments for ArraysZip", - expr.children ++ expr.names: _*) + withFallbackReason(expr, "unsupported arguments for ArraysZip") None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/bitwise.scala b/spark/src/main/scala/org/apache/comet/serde/bitwise.scala index db773cd541..115bd80422 100644 --- a/spark/src/main/scala/org/apache/comet/serde/bitwise.scala +++ b/spark/src/main/scala/org/apache/comet/serde/bitwise.scala @@ -53,7 +53,7 @@ object CometBitwiseNot extends CometExpressionSerde[BitwiseNot] { val childProto = exprToProto(expr.child, inputs, binding) val bitNotScalarExpr = scalarFunctionExprToProto("bitwise_not", childProto) - optExprWithFallbackReason(bitNotScalarExpr, expr, expr.children: _*) + bitNotScalarExpr } } @@ -148,7 +148,7 @@ object CometBitwiseGet extends CometExpressionSerde[BitwiseGet] { val posProto = exprToProto(expr.right, inputs, binding) val bitGetScalarExpr = scalarFunctionExprToProtoWithReturnType("bit_get", ByteType, false, argProto, posProto) - optExprWithFallbackReason(bitGetScalarExpr, expr, expr.children: _*) + bitGetScalarExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/conditional.scala b/spark/src/main/scala/org/apache/comet/serde/conditional.scala index cd07730890..df94be8c93 100644 --- a/spark/src/main/scala/org/apache/comet/serde/conditional.scala +++ b/spark/src/main/scala/org/apache/comet/serde/conditional.scala @@ -23,7 +23,6 @@ import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{Attribute, CaseWhen, Coalesce, Expression, If, IsNotNull} -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal object CometIf extends CometExpressionSerde[If] { @@ -45,7 +44,6 @@ object CometIf extends CometExpressionSerde[If] { .setIf(builder) .build()) } else { - withFallbackReason(expr, expr.predicate, expr.trueValue, expr.falseValue) None } } @@ -76,7 +74,6 @@ object CometCaseWhen extends CometExpressionSerde[CaseWhen] { if (elseValueExpr.isDefined) { builder.setElseExpr(elseValueExpr.get) } else { - withFallbackReason(expr, expr.elseValue.get) return None } } @@ -86,7 +83,6 @@ object CometCaseWhen extends CometExpressionSerde[CaseWhen] { .setCaseWhen(builder) .build()) } else { - withFallbackReason(expr, allBranches: _*) None } } @@ -116,7 +112,6 @@ object CometCoalesce extends CometExpressionSerde[Coalesce] { if (elseValueExpr.isDefined) { builder.setElseExpr(elseValueExpr.get) } else { - withFallbackReason(expr, elseValue) return None } Some( @@ -125,7 +120,6 @@ object CometCoalesce extends CometExpressionSerde[Coalesce] { .setCaseWhen(builder) .build()) } else { - withFallbackReason(expr, branches.map(_._2): _*) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala b/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala index 9b30357f9f..460f23485d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala +++ b/spark/src/main/scala/org/apache/comet/serde/contraintExpressions.scala @@ -22,7 +22,7 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, KnownFloatingPointNormalized, KnownNullable} import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, serializeDataType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, serializeDataType} object CometKnownFloatingPointNormalized extends CometExpressionSerde[KnownFloatingPointNormalized] { @@ -63,7 +63,7 @@ object CometKnownFloatingPointNormalized .setDatatype(dataType) ExprOuterClass.Expr.newBuilder().setNormalizeNanAndZero(builder).build() } - optExprWithFallbackReason(optExpr, expr, wrapped) + optExpr case child => // Nested normalization (array / struct / map). Spark 4.2 normalizes the inputs to @@ -72,7 +72,7 @@ object CometKnownFloatingPointNormalized // `KnownFloatingPointNormalized` is a runtime no-op tag, so serialize the child directly // and let its serde (e.g. the ArrayTransform codegen dispatcher) carry the normalization. val optExpr = exprToProtoInternal(child, inputs, binding) - optExprWithFallbackReason(optExpr, expr, child) + optExpr } } } @@ -89,6 +89,6 @@ object CometKnownNullable extends CometExpressionSerde[KnownNullable] { inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { val optExpr = exprToProtoInternal(expr.child, inputs, binding) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/datetime.scala b/spark/src/main/scala/org/apache/comet/serde/datetime.scala index a2600bf688..abb39a8ea7 100644 --- a/spark/src/main/scala/org/apache/comet/serde/datetime.scala +++ b/spark/src/main/scala/org/apache/comet/serde/datetime.scala @@ -79,7 +79,7 @@ trait CometExprGetDateField[T <: GetDateField] { .build()) .build() }) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -137,7 +137,7 @@ object CometDayOfWeek .build() } .headOption - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -166,7 +166,7 @@ object CometWeekDay extends CometExpressionSerde[WeekDay] with CometExprGetDateF .build() } .headOption - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -222,7 +222,6 @@ object CometHour extends CometExpressionSerde[Hour] { .setHour(builder) .build()) } else { - withFallbackReason(expr, expr.child) None } } @@ -249,7 +248,6 @@ object CometMinute extends CometExpressionSerde[Minute] { .setMinute(builder) .build()) } else { - withFallbackReason(expr, expr.child) None } } @@ -276,7 +274,6 @@ object CometSecond extends CometExpressionSerde[Second] { .setSecond(builder) .build()) } else { - withFallbackReason(expr, expr.child) None } } @@ -348,7 +345,6 @@ object CometUnixTimestamp extends CometExpressionSerde[UnixTimestamp] { .setUnixTimestamp(builder) .build()) } else { - withFallbackReason(expr, expr.children.head) None } } @@ -383,7 +379,7 @@ object CometFromUTCTimestamp binding: Boolean): Option[ExprOuterClass.Expr] = { val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProto("from_utc_timestamp", childExprs: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -403,7 +399,7 @@ object CometToUTCTimestamp binding: Boolean): Option[ExprOuterClass.Expr] = { val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProto("to_utc_timestamp", childExprs: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -434,7 +430,7 @@ object CometConvertTimezone val ts = exprToProtoInternal(expr.sourceTs, inputs, binding) val toUtc = scalarFunctionExprToProto("to_utc_timestamp", ts, srcTz) val fromUtc = scalarFunctionExprToProto("from_utc_timestamp", toUtc, tgtTz) - optExprWithFallbackReason(fromUtc, expr, expr.children: _*) + fromUtc } } @@ -470,7 +466,7 @@ object CometNextDay extends CometExpressionSerde[NextDay] { DateType, expr.failOnError, childExpr: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -496,7 +492,7 @@ object CometMakeDate extends CometExpressionSerde[MakeDate] { DateType, expr.failOnError, childExpr: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -543,7 +539,7 @@ object CometUnixDate extends CometExpressionSerde[UnixDate] { .build()) .build() } - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -597,7 +593,7 @@ object CometTruncDate extends CometExpressionSerde[TruncDate] with CodegenDispat false, childExpr, formatExpr) - optExprWithFallbackReason(optExpr, expr, expr.date, expr.format) + optExpr } } @@ -687,7 +683,6 @@ object CometTruncTimestamp .setTruncTimestamp(builder) .build()) } else { - withFallbackReason(expr, expr.timestamp, expr.format) None } } @@ -811,7 +806,7 @@ object CometDateFormat false, childExpr, formatExpr) - optExprWithFallbackReason(optExpr, expr, expr.left, expr.right) + optExpr } else { // Hand the full `DateFormatClass` (with `timeZoneId` already stamped by `ResolveTimeZone`) // to the codegen dispatcher. It closure-serializes the bound tree, so non-UTC timezones @@ -851,7 +846,7 @@ object CometHours extends CometExpressionSerde[Hours] { .setHoursTransform(builder) .build() } - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -908,7 +903,7 @@ object CometDays extends CometExpressionSerde[Days] { .build() } - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -1017,6 +1012,6 @@ object CometPreciseTimestampConversion extends CometExpressionSerde[PreciseTimes .setDatatype(dataType) ExprOuterClass.Expr.newBuilder().setPreciseTimestampConversion(builder).build() } - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/decimalExpressions.scala b/spark/src/main/scala/org/apache/comet/serde/decimalExpressions.scala index 3af32e1d11..08e632cc77 100644 --- a/spark/src/main/scala/org/apache/comet/serde/decimalExpressions.scala +++ b/spark/src/main/scala/org/apache/comet/serde/decimalExpressions.scala @@ -22,7 +22,7 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, MakeDecimal, UnscaledValue} import org.apache.spark.sql.types.{DecimalType, LongType} -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProtoWithReturnType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProtoWithReturnType} object CometUnscaledValue extends CometExpressionSerde[UnscaledValue] { override def convert( @@ -32,7 +32,7 @@ object CometUnscaledValue extends CometExpressionSerde[UnscaledValue] { val childExpr = exprToProtoInternal(expr.child, inputs, binding) val optExpr = scalarFunctionExprToProtoWithReturnType("unscaled_value", LongType, false, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -66,7 +66,7 @@ object CometMakeDecimal extends CometExpressionSerde[MakeDecimal] { DecimalType(expr.precision, expr.scale), failOnError = !expr.nullOnOverflow, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/json.scala b/spark/src/main/scala/org/apache/comet/serde/json.scala index 02a9e14eb5..3a486ce71a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/json.scala +++ b/spark/src/main/scala/org/apache/comet/serde/json.scala @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, JsonObjectKeys, Len import org.apache.comet.CometConf import org.apache.comet.serde.ExprOuterClass.Expr -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto} /** * `json_array_length` runs Spark's own implementation through the codegen dispatcher by default, @@ -56,7 +56,7 @@ object CometLengthOfJsonArray if (CometConf.isExprAllowIncompat(getExprConfigName(expr))) { val childExpr = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProto("json_array_length", childExpr: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } else { super.convert(expr, inputs, binding) } diff --git a/spark/src/main/scala/org/apache/comet/serde/maps.scala b/spark/src/main/scala/org/apache/comet/serde/maps.scala index 817792e1d2..c889284129 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ -import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} +import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, scalarFunctionExprToProto} import org.apache.comet.shims.CometTypeShim object CometMapKeys extends CometExpressionSerde[MapKeys] { @@ -34,7 +34,7 @@ object CometMapKeys extends CometExpressionSerde[MapKeys] { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(expr.child, inputs, binding) val mapKeysScalarExpr = scalarFunctionExprToProto("map_keys", childExpr) - optExprWithFallbackReason(mapKeysScalarExpr, expr, expr.children: _*) + mapKeysScalarExpr } } @@ -46,7 +46,7 @@ object CometMapEntries extends CometExpressionSerde[MapEntries] { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(expr.child, inputs, binding) val mapEntriesScalarExpr = scalarFunctionExprToProto("map_entries", childExpr) - optExprWithFallbackReason(mapEntriesScalarExpr, expr, expr.children: _*) + mapEntriesScalarExpr } } @@ -58,7 +58,7 @@ object CometMapValues extends CometExpressionSerde[MapValues] { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(expr.child, inputs, binding) val mapValuesScalarExpr = scalarFunctionExprToProto("map_values", childExpr) - optExprWithFallbackReason(mapValuesScalarExpr, expr, expr.children: _*) + mapValuesScalarExpr } } @@ -71,7 +71,7 @@ object CometMapExtract extends CometExpressionSerde[GetMapValue] { val mapExpr = exprToProtoInternal(expr.child, inputs, binding) val keyExpr = exprToProtoInternal(expr.key, inputs, binding) val mapExtractExpr = scalarFunctionExprToProto("map_extract", mapExpr, keyExpr) - optExprWithFallbackReason(mapExtractExpr, expr, expr.children: _*) + mapExtractExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/math.scala b/spark/src/main/scala/org/apache/comet/serde/math.scala index e70794faa3..a0ea76cc06 100644 --- a/spark/src/main/scala/org/apache/comet/serde/math.scala +++ b/spark/src/main/scala/org/apache/comet/serde/math.scala @@ -22,8 +22,7 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Abs, Add, Atan2, Attribute, BRound, Ceil, CheckOverflow, Conv, Expression, Floor, Hex, Hypot, If, LessThanOrEqual, Literal, Log, Log10, Log1p, Log2, Logarithm, NaNvl, Pmod, Pow, UnaryPositive, Unhex, WidthBucket} import org.apache.spark.sql.types.{DecimalType, DoubleType, NumericType} -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType, serializeDataType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType, serializeDataType} object CometAtan2 extends CometExpressionSerde[Atan2] { override def convert( @@ -36,7 +35,7 @@ object CometAtan2 extends CometExpressionSerde[Atan2] { val leftExpr = exprToProtoInternal(left, inputs, binding) val rightExpr = exprToProtoInternal(right, inputs, binding) val optExpr = scalarFunctionExprToProto("atan2", leftExpr, rightExpr) - optExprWithFallbackReason(optExpr, expr, expr.left, expr.right) + optExpr } } @@ -59,7 +58,7 @@ object CometCeil extends CometExpressionSerde[Ceil] { case _ => val optExpr = scalarFunctionExprToProtoWithReturnType("ceil", expr.dataType, false, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } } @@ -83,7 +82,7 @@ object CometFloor extends CometExpressionSerde[Floor] { case _ => val optExpr = scalarFunctionExprToProtoWithReturnType("floor", expr.dataType, false, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } } @@ -98,7 +97,7 @@ object CometLog extends CometExpressionSerde[Log] with MathExprBase { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(nullIfNegative(expr.child), inputs, binding) val optExpr = scalarFunctionExprToProto("ln", childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -109,7 +108,7 @@ object CometLog10 extends CometExpressionSerde[Log10] with MathExprBase { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(nullIfNegative(expr.child), inputs, binding) val optExpr = scalarFunctionExprToProto("log10", childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -120,7 +119,7 @@ object CometLog2 extends CometExpressionSerde[Log2] with MathExprBase { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(nullIfNegative(expr.child), inputs, binding) val optExpr = scalarFunctionExprToProto("log2", childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -136,7 +135,7 @@ object CometLogarithm extends CometExpressionSerde[Logarithm] { val rightExpr = exprToProtoInternal(expr.right, inputs, binding) val optExpr = scalarFunctionExprToProtoWithReturnType("spark_log", DoubleType, false, leftExpr, rightExpr) - optExprWithFallbackReason(optExpr, expr, expr.left, expr.right) + optExpr } } @@ -147,7 +146,7 @@ object CometHex extends CometExpressionSerde[Hex] with MathExprBase { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(expr.child, inputs, binding) val optExpr = scalarFunctionExprToProtoWithReturnType("hex", expr.dataType, false, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -166,7 +165,7 @@ object CometUnhex extends CometExpressionSerde[Unhex] with MathExprBase { false, childExpr, failOnErrorExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -200,7 +199,7 @@ object CometAbs extends CometExpressionSerde[Abs] with MathExprBase { false, childExpr, failOnErrorExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -213,7 +212,7 @@ object CometPow extends CometExpressionSerde[Pow] { val leftExpr = exprToProtoInternal(expr.left, inputs, binding) val rightExpr = exprToProtoInternal(expr.right, inputs, binding) val optExpr = scalarFunctionExprToProto("pow", leftExpr, rightExpr) - optExprWithFallbackReason(optExpr, expr, expr.left, expr.right) + optExpr } } @@ -258,7 +257,6 @@ object CometCheckOverflow extends CometExpressionSerde[CheckOverflow] { .setCheckOverflow(builder) .build()) } else { - withFallbackReason(expr, expr.child) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala b/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala index fb910ec3ac..62c2e1eead 100644 --- a/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala +++ b/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala @@ -30,9 +30,7 @@ object CometAlias extends CometExpressionSerde[Alias] { inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { val r = exprToProtoInternal(a.child, inputs, binding) - if (r.isEmpty) { - withFallbackReason(a, a.child) - } + if (r.isEmpty) {} r } } diff --git a/spark/src/main/scala/org/apache/comet/serde/predicates.scala b/spark/src/main/scala/org/apache/comet/serde/predicates.scala index af865b410c..7ae1c35940 100644 --- a/spark/src/main/scala/org/apache/comet/serde/predicates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/predicates.scala @@ -26,7 +26,7 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.BooleanType import org.apache.comet.CometConf -import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, withFallbackReason} +import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus} import org.apache.comet.serde.ExprOuterClass.Expr import org.apache.comet.serde.QueryPlanSerde._ @@ -244,7 +244,7 @@ object CometIsNaN extends CometExpressionSerde[IsNaN] { val childExpr = exprToProtoInternal(expr.child, inputs, binding) val optExpr = scalarFunctionExprToProtoWithReturnType("isnan", BooleanType, false, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } @@ -386,7 +386,6 @@ object ComparisonUtils { .build()) } else { val allExprs = list ++ Seq(value) - withFallbackReason(expr, allExprs: _*) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/statics.scala b/spark/src/main/scala/org/apache/comet/serde/statics.scala index f415c373a1..6501a1d46c 100644 --- a/spark/src/main/scala/org/apache/comet/serde/statics.scala +++ b/spark/src/main/scala/org/apache/comet/serde/statics.scala @@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.util.CharVarcharCodegenUtils import org.apache.spark.sql.types.StringType import org.apache.comet.CometSparkSessionExtensions.withFallbackReason -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} object CometStaticInvoke extends CometExpressionSerde[StaticInvoke] { @@ -61,8 +61,7 @@ object CometStaticInvoke extends CometExpressionSerde[StaticInvoke] { case None => withFallbackReason( expr, - s"Static invoke expression: ${expr.functionName} is not supported", - expr.children: _*) + s"Static invoke expression: ${expr.functionName} is not supported") None } } @@ -75,7 +74,7 @@ object CometUrlEncodeStaticInvoke extends CometExpressionSerde[StaticInvoke] { binding: Boolean): Option[ExprOuterClass.Expr] = { val childExpr = exprToProtoInternal(expr.children.head, inputs, binding) val optExpr = scalarFunctionExprToProto("url_encode", childExpr) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -91,7 +90,7 @@ object CometUrlDecodeStaticInvoke extends CometExpressionSerde[StaticInvoke] { val funcName = if (failOnError) "url_decode" else "try_url_decode" val childExpr = exprToProtoInternal(expr.children.head, inputs, binding) val optExpr = scalarFunctionExprToProto(funcName, childExpr) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -113,7 +112,7 @@ object CometBase64StaticInvoke extends CometExpressionSerde[StaticInvoke] { failOnError = false, childExpr, chunkExpr) - optExprWithFallbackReason(optExpr, expr, expr.arguments: _*) + optExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index 7422dadd97..ebf4508988 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -24,7 +24,7 @@ import org.apache.spark.sql.types.{BinaryType, DataTypes, IntegerType, LongType, import org.apache.comet.CometConf import org.apache.comet.serde.ExprOuterClass.Expr -import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} +import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} import org.apache.comet.shims.CometTypeShim object CometStringRepeat extends CometExpressionSerde[StringRepeat] { @@ -43,7 +43,7 @@ object CometStringRepeat extends CometExpressionSerde[StringRepeat] { val leftExpr = exprToProtoInternal(leftCast, inputs, binding) val rightExpr = exprToProtoInternal(rightCast, inputs, binding) val optExpr = scalarFunctionExprToProto("repeat", leftExpr, rightExpr) - optExprWithFallbackReason(optExpr, expr, leftCast, rightCast) + optExpr } } @@ -139,7 +139,7 @@ object CometLevenshtein extends CometExpressionSerde[Levenshtein] { val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProtoWithReturnType("levenshtein", IntegerType, false, childExprs: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } @@ -220,7 +220,7 @@ object CometSubstringIndex extends CometExpressionSerde[SubstringIndex] { val countExpr = exprToProtoInternal(countCast, inputs, binding) val optExpr = scalarFunctionExprToProto("substring_index", strExpr, delimExpr, countExpr) - optExprWithFallbackReason(optExpr, expr, expr.strExpr, expr.delimExpr, expr.countExpr) + optExpr } } @@ -489,7 +489,7 @@ object CometRegExpExtract extends CometExpressionSerde[RegExpExtract] { subjectExpr, patternExpr, idxExpr) - optExprWithFallbackReason(optExpr, expr, expr.subject, expr.regexp, expr.idx) + optExpr } else { // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the // Comet pipeline. Falls back to Spark when the dispatcher is disabled. @@ -527,7 +527,7 @@ object CometRegExpExtractAll extends CometExpressionSerde[RegExpExtractAll] { subjectExpr, patternExpr, idxExpr) - optExprWithFallbackReason(optExpr, expr, expr.subject, expr.regexp, expr.idx) + optExpr } else { // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the // Comet pipeline. Falls back to Spark when the dispatcher is disabled. @@ -577,7 +577,7 @@ object CometRegExpReplace extends CometExpressionSerde[RegExpReplace] with Nativ patternExpr, replacementExpr, flagsExpr) - optExprWithFallbackReason(optExpr, expr, expr.subject, expr.regexp, expr.rep, expr.pos) + optExpr } else { // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the // Comet pipeline. Falls back to Spark when the dispatcher is disabled. @@ -623,7 +623,7 @@ object CometStringSplit extends CometExpressionSerde[StringSplit] with NativeOpt strExpr, regexExpr, limitExpr) - optExprWithFallbackReason(optExpr, expr, expr.str, expr.regex, expr.limit) + optExpr } else { // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the // Comet pipeline. Falls back to Spark when the dispatcher is disabled. @@ -673,7 +673,7 @@ object CometGetJsonObject extends CometCodegenDispatch[GetJsonObject] with Nativ false, jsonExpr, pathExpr) - optExprWithFallbackReason(optExpr, expr, expr.json, expr.path) + optExpr } else { super.convert(expr, inputs, binding) } @@ -709,7 +709,7 @@ object CometBase64 extends CometExpressionSerde[Base64] { failOnError = false, childExpr, chunkExpr) - optExprWithFallbackReason(optExpr, expr, expr.child) + optExpr } } diff --git a/spark/src/main/scala/org/apache/comet/serde/structs.scala b/spark/src/main/scala/org/apache/comet/serde/structs.scala index 409ef38b4f..2f9619d491 100644 --- a/spark/src/main/scala/org/apache/comet/serde/structs.scala +++ b/spark/src/main/scala/org/apache/comet/serde/structs.scala @@ -63,7 +63,7 @@ object CometCreateNamedStruct extends CometExpressionSerde[CreateNamedStruct] { .setCreateNamedStruct(structBuilder) .build()) } else { - withFallbackReason(expr, "unsupported arguments for CreateNamedStruct", expr.valExprs: _*) + withFallbackReason(expr, "unsupported arguments for CreateNamedStruct") None } @@ -108,7 +108,7 @@ object CometGetArrayStructFields extends CometExpressionSerde[GetArrayStructFiel .setGetArrayStructFields(arrayStructFieldsBuilder) .build()) } else { - withFallbackReason(expr, "unsupported arguments for GetArrayStructFields", expr.child) + withFallbackReason(expr, "unsupported arguments for GetArrayStructFields") None } } @@ -160,7 +160,6 @@ object CometStructsToJson extends CometCodegenDispatch[StructsToJson] with Nativ .setToJson(toJson) .build()) case _ => - withFallbackReason(expr, expr.child) None } } else { diff --git a/spark/src/main/scala/org/apache/comet/serde/unixtime.scala b/spark/src/main/scala/org/apache/comet/serde/unixtime.scala index f88528c468..08285c2cf1 100644 --- a/spark/src/main/scala/org/apache/comet/serde/unixtime.scala +++ b/spark/src/main/scala/org/apache/comet/serde/unixtime.scala @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, FromUnixTime, Liter import org.apache.spark.sql.catalyst.util.TimestampFormatter import org.apache.comet.CometSparkSessionExtensions.withFallbackReason -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto} // TODO: DataFusion supports only -8334601211038 <= sec <= 8210266876799 // https://github.com/apache/datafusion/issues/16594 @@ -80,9 +80,8 @@ object CometFromUnixTime extends CometExpressionSerde[FromUnixTime] with Codegen val timestampExpr = scalarFunctionExprToProto("from_unixtime", Seq(secExpr, timeZone): _*) val optExpr = scalarFunctionExprToProto("to_char", Seq(timestampExpr, formatExpr): _*) - optExprWithFallbackReason(optExpr, expr, expr.sec, expr.format) + optExpr } else { - withFallbackReason(expr, expr.sec, expr.format) None } } diff --git a/spark/src/main/scala/org/apache/comet/serde/url.scala b/spark/src/main/scala/org/apache/comet/serde/url.scala index b54d78ba25..891d12f333 100644 --- a/spark/src/main/scala/org/apache/comet/serde/url.scala +++ b/spark/src/main/scala/org/apache/comet/serde/url.scala @@ -21,7 +21,7 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, ParseUrl} -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto} object CometParseUrl extends CometExpressionSerde[ParseUrl] { @@ -32,6 +32,6 @@ object CometParseUrl extends CometExpressionSerde[ParseUrl] { val funcName = if (expr.failOnError) "parse_url" else "try_parse_url" val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProto(funcName, childExprs: _*) - optExprWithFallbackReason(optExpr, expr, expr.children: _*) + optExpr } } diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala index 97ef6417d6..4d4abf3d2e 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala @@ -51,7 +51,7 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { val winExprs: Seq[WindowExpressionInfo] = op.windowExpression.map { expr => extractWindowExpression(expr).getOrElse { - withFallbackReason(op, s"Unsupported window expression: $expr", expr) + withFallbackReason(op, s"Unsupported window expression: $expr") return None } } @@ -74,16 +74,8 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { windowBuilder.addAllOrderByList(sortOrders.map(_.get).asJava) Some(builder.setWindow(windowBuilder).build()) } else { - // Roll up reasons already attached to per-expression nodes so the Window - // operator itself carries a fallback attribution. Without this, the plan - // prints a bare `Window` and the real reason lives on a sub-expression - // that isn't obvious in the standard explain output. - val failing = winExprs.toSeq.zip(windowExprProto).collect { case (we, None) => - we.windowExpression - } ++ - op.partitionSpec.zip(partitionExprs).collect { case (e, None) => e } ++ - op.orderSpec.zip(sortOrders).collect { case (e, None) => e } - withFallbackReason(op, failing: _*) + // Whichever of the window / partition / order expressions failed has already recorded its + // own reason; `CometExecRule.rollUpFallbackReasons` lifts it onto this operator. None } } @@ -204,28 +196,28 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { if (AggSerde.minMaxDataTypeSupported(min.dataType)) { Some(agg) } else { - withFallbackReason(windowExpr, s"datatype ${min.dataType} is not supported", expr) + withFallbackReason(windowExpr, s"datatype ${min.dataType} is not supported") None } case max: Max => if (AggSerde.minMaxDataTypeSupported(max.dataType)) { Some(agg) } else { - withFallbackReason(windowExpr, s"datatype ${max.dataType} is not supported", expr) + withFallbackReason(windowExpr, s"datatype ${max.dataType} is not supported") None } case s: Sum => if (AggSerde.sumDataTypeSupported(s.dataType)) { Some(agg) } else { - withFallbackReason(windowExpr, s"datatype ${s.dataType} is not supported", expr) + withFallbackReason(windowExpr, s"datatype ${s.dataType} is not supported") None } case a: Average => if (AggSerde.avgDataTypeSupported(a.dataType)) { Some(agg) } else { - withFallbackReason(windowExpr, s"datatype ${a.dataType} is not supported", expr) + withFallbackReason(windowExpr, s"datatype ${a.dataType} is not supported") None } case _: First => @@ -236,8 +228,7 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { withFallbackReason( windowExpr, s"aggregate ${agg.aggregateFunction}" + - " is not supported for window function", - expr) + " is not supported for window function") None } case _ => @@ -268,7 +259,7 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { windowExpr.windowFunction match { case lag: Lag if !lag.default.isInstanceOf[Literal] => // https://github.com/apache/datafusion-comet/issues/4268 - withFallbackReason(windowExpr, "Lag default value must be a literal", lag.default) + withFallbackReason(windowExpr, "Lag default value must be a literal") (None, None, false) case lag: Lag => val inputExpr = exprToProto(lag.input, output) @@ -278,7 +269,7 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { (None, func, lag.ignoreNulls) case lead: Lead if !lead.default.isInstanceOf[Literal] => // https://github.com/apache/datafusion-comet/issues/4268 - withFallbackReason(windowExpr, "Lead default value must be a literal", lead.default) + withFallbackReason(windowExpr, "Lead default value must be a literal") (None, None, false) case lead: Lead => val inputExpr = exprToProto(lead.input, output) @@ -317,8 +308,7 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] { case other => withFallbackReason( windowExpr, - s"window function ${other.getClass.getSimpleName} is not supported", - other) + s"window function ${other.getClass.getSimpleName} is not supported") (None, None, false) } } diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index 9c4917e2d6..b48a1eb5cc 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala @@ -1039,7 +1039,6 @@ object CometProjectExec extends CometOperatorSerde[ProjectExec] { .addAllProjectList(exprs.map(_.get).asJava) Some(builder.setProjection(projectBuilder).build()) } else { - withFallbackReason(op, op.projectList: _*) None } } @@ -1099,7 +1098,6 @@ object CometFilterExec extends CometOperatorSerde[FilterExec] { .setPredicate(cond.get) Some(builder.setFilter(filterBuilder).build()) } else { - withFallbackReason(op, op.condition, op.child) None } } @@ -1172,7 +1170,7 @@ object CometSortExec extends CometOperatorSerde[SortExec] { .addAllSortOrders(sortOrders.map(_.get).asJava) Some(builder.setSort(sortBuilder).build()) } else { - withFallbackReason(op, "sort order not supported", op.sortOrder: _*) + withFallbackReason(op, "sort order not supported") None } } @@ -1362,11 +1360,7 @@ object CometExpandExec extends CometOperatorSerde[ExpandExec] { op: ExpandExec, builder: Operator.Builder, childOp: OperatorOuterClass.Operator*): Option[OperatorOuterClass.Operator] = { - var allProjExprs: Seq[Expression] = Seq() - val projExprs = op.projections.flatMap(_.map(e => { - allProjExprs = allProjExprs :+ e - exprToProto(e, op.child.output) - })) + val projExprs = op.projections.flatMap(_.map(e => exprToProto(e, op.child.output))) if (projExprs.forall(_.isDefined) && childOp.nonEmpty) { val expandBuilder = OperatorOuterClass.Expand @@ -1375,7 +1369,6 @@ object CometExpandExec extends CometOperatorSerde[ExpandExec] { .setNumExprPerProject(op.projections.head.size) Some(builder.setExpand(expandBuilder).build()) } else { - withFallbackReason(op, allProjExprs: _*) None } } @@ -1460,7 +1453,6 @@ object CometExplodeExec extends CometOperatorSerde[GenerateExec] { val childExprProto = exprToProto(childExpr, op.child.output) if (childExprProto.isEmpty) { - withFallbackReason(op, childExpr) return None } @@ -1472,7 +1464,6 @@ object CometExplodeExec extends CometOperatorSerde[GenerateExec] { } if (projectExprs.exists(_.isEmpty) || childOp.isEmpty) { - withFallbackReason(op, op.output: _*) return None } @@ -1792,10 +1783,7 @@ trait CometBaseAggregate { } if (aggExprs.exists(_.isEmpty)) { - withFallbackReason( - aggregate, - "Unsupported aggregate expression(s)", - aggregateExpressions ++ aggregateExpressions.map(_.aggregateFunction): _*) + withFallbackReason(aggregate, "Unsupported aggregate expression(s)") return None } @@ -1836,9 +1824,6 @@ trait CometBaseAggregate { Some(builder.setHashAgg(hashAggBuilder).build()) } } else { - val allChildren: Seq[Expression] = - groupingExpressions ++ aggregateExpressions ++ aggregateAttributes - withFallbackReason(aggregate, allChildren: _*) None } } @@ -1867,8 +1852,7 @@ trait CometBaseAggregate { if (resultExprs.exists(_.isEmpty)) { withFallbackReason( aggregate, - s"Unsupported result expressions found in: $resultExpressions", - resultExpressions: _*) + s"Unsupported result expressions found in: $resultExpressions") return None } val planId = builder.getPlanId @@ -2151,7 +2135,6 @@ trait CometHashJoin { val condition = join.condition.map { cond => val condProto = exprToProto(cond, join.left.output ++ join.right.output) if (condProto.isEmpty) { - withFallbackReason(join, cond) return None } condProto.get @@ -2190,8 +2173,6 @@ trait CometHashJoin { condition.foreach(joinBuilder.setCondition) Some(builder.setHashJoin(joinBuilder).build()) } else { - val allExprs: Seq[Expression] = joinKeys - withFallbackReason(join, allExprs: _*) None } } @@ -2318,7 +2299,6 @@ object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNest val joinCondition = op.condition.map({ cond => val condProto = exprToProto(cond, op.left.output ++ op.right.output) if (condProto.isEmpty) { - withFallbackReason(op, cond) return None } condProto.get @@ -2636,15 +2616,13 @@ object CometSortMergeJoinExec extends CometOperatorSerde[SortMergeJoinExec] { .get(join.conf)) { withFallbackReason( join, - s"${CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED.key} is not enabled", - join.condition.get) + s"${CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED.key} is not enabled") return None } val condition = join.condition.map { cond => val condProto = exprToProto(cond, join.left.output ++ join.right.output) if (condProto.isEmpty) { - withFallbackReason(join, cond) return None } condProto.get @@ -2705,8 +2683,6 @@ object CometSortMergeJoinExec extends CometOperatorSerde[SortMergeJoinExec] { condition.map(joinBuilder.setCondition) Some(builder.setSortMergeJoin(joinBuilder).build()) } else { - val allExprs: Seq[Expression] = joinKeys - withFallbackReason(join, allExprs: _*) None } } diff --git a/spark/src/main/spark-3.5/org/apache/comet/serde/CometToPrettyString.scala b/spark/src/main/spark-3.5/org/apache/comet/serde/CometToPrettyString.scala index 9926102553..147eeb1e29 100644 --- a/spark/src/main/spark-3.5/org/apache/comet/serde/CometToPrettyString.scala +++ b/spark/src/main/spark-3.5/org/apache/comet/serde/CometToPrettyString.scala @@ -22,7 +22,6 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, ToPrettyString} import org.apache.spark.sql.types.DataTypes -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.expressions.{CometCast, CometEvalMode} import org.apache.comet.serde.QueryPlanSerde.{binaryOutputStyle, exprToProtoInternal} @@ -57,7 +56,6 @@ object CometToPrettyString extends CometExpressionSerde[ToPrettyString] { .build() Some(ExprOuterClass.Expr.newBuilder().setToPrettyString(tps).build()) case _ => - withFallbackReason(expr, expr.child) None } } diff --git a/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala b/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala index 64474291e7..a2380e4008 100644 --- a/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala +++ b/spark/src/main/spark-4.1+/org/apache/comet/shims/CometExprShim.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.types.TimeType import org.apache.comet.expressions.CometEvalMode import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr} -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProtoWithReturnType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProtoWithReturnType} /** * `CometExprShim` acts as a shim for parsing expressions from different Spark versions. @@ -64,7 +64,7 @@ trait CometExprShim extends Spark4xCometExprShim { val childExprs = s.arguments.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProtoWithReturnType("make_time", s.dataType, true, childExprs: _*) - optExprWithFallbackReason(optExpr, expr, s.arguments: _*) + optExpr case i: Invoke => (i.targetObject, i.functionName, i.arguments) match { @@ -73,7 +73,7 @@ trait CometExprShim extends Spark4xCometExprShim { val childExprs = args.map(exprToProtoInternal(_, inputs, binding)) val optExpr = scalarFunctionExprToProtoWithReturnType("to_time", i.dataType, true, childExprs: _*) - optExprWithFallbackReason(optExpr, i, args: _*) + optExpr case _ => super.sparkVersionSpecificExprToProtoInternal(expr, inputs, binding) } @@ -89,7 +89,7 @@ trait CometExprShim extends Spark4xCometExprShim { i.dataType, false, childExprs: _*) - optExprWithFallbackReason(optExpr, expr, args: _*) + optExpr case _ => super.sparkVersionSpecificExprToProtoInternal(expr, inputs, binding) } diff --git a/spark/src/main/spark-4.x/org/apache/comet/serde/CometMapSort.scala b/spark/src/main/spark-4.x/org/apache/comet/serde/CometMapSort.scala index 341c8fc24e..4fad289a08 100644 --- a/spark/src/main/spark-4.x/org/apache/comet/serde/CometMapSort.scala +++ b/spark/src/main/spark-4.x/org/apache/comet/serde/CometMapSort.scala @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, MapSort} import org.apache.spark.sql.types.MapType import org.apache.comet.CometConf -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProtoWithReturnType, supportedScalarSortElementType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProtoWithReturnType, supportedScalarSortElementType} object CometMapSort extends CometExpressionSerde[MapSort] { @@ -57,6 +57,6 @@ object CometMapSort extends CometExpressionSerde[MapSort] { expr.dataType, failOnError = false, childExpr) - optExprWithFallbackReason(mapSortExpr, expr, expr.child) + mapSortExpr } } diff --git a/spark/src/main/spark-4.x/org/apache/comet/serde/CometToPrettyString.scala b/spark/src/main/spark-4.x/org/apache/comet/serde/CometToPrettyString.scala index 9926102553..147eeb1e29 100644 --- a/spark/src/main/spark-4.x/org/apache/comet/serde/CometToPrettyString.scala +++ b/spark/src/main/spark-4.x/org/apache/comet/serde/CometToPrettyString.scala @@ -22,7 +22,6 @@ package org.apache.comet.serde import org.apache.spark.sql.catalyst.expressions.{Attribute, ToPrettyString} import org.apache.spark.sql.types.DataTypes -import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.expressions.{CometCast, CometEvalMode} import org.apache.comet.serde.QueryPlanSerde.{binaryOutputStyle, exprToProtoInternal} @@ -57,7 +56,6 @@ object CometToPrettyString extends CometExpressionSerde[ToPrettyString] { .build() Some(ExprOuterClass.Expr.newBuilder().setToPrettyString(tps).build()) case _ => - withFallbackReason(expr, expr.child) None } } diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/CometExprShim4x.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/CometExprShim4x.scala index 234be4dc54..e15db29660 100644 --- a/spark/src/main/spark-4.x/org/apache/comet/shims/CometExprShim4x.scala +++ b/spark/src/main/spark-4.x/org/apache/comet/shims/CometExprShim4x.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.expressions.xml.{XmlExpressionEvalUtils, XP import org.apache.comet.CometSparkSessionExtensions.withFallbackReason import org.apache.comet.serde.CometScalaUDF import org.apache.comet.serde.ExprOuterClass.Expr -import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, hasNonDefaultStringCollation, optExprWithFallbackReason, scalarFunctionExprToProtoWithReturnType} +import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, hasNonDefaultStringCollation, scalarFunctionExprToProtoWithReturnType} /** * Expression conversions shared across all Spark 4.x minor versions, compiled from the @@ -58,12 +58,12 @@ trait CometExprShim4x { val childExpr = exprToProtoInternal(d.child, inputs, binding) val nameExpr = scalarFunctionExprToProtoWithReturnType("dayname", d.dataType, false, childExpr) - optExprWithFallbackReason(nameExpr, d, d.child) + nameExpr case m: MonthName => val childExpr = exprToProtoInternal(m.child, inputs, binding) val nameExpr = scalarFunctionExprToProtoWithReturnType("monthname", m.dataType, false, childExpr) - optExprWithFallbackReason(nameExpr, m, m.child) + nameExpr case _ => None } @@ -90,7 +90,7 @@ trait CometExprShim4x { false, strExpr, delimiterExpr) - optExprWithFallbackReason(splitExpr, expr, expr.str, expr.delimiter) + splitExpr } // Spark 4.x lowers the RuntimeReplaceable structured-text functions to an evaluator-backed diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala index 8091d9dd0a..50161d46c7 100644 --- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala @@ -1014,14 +1014,12 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { val query = sql(s"select cast(id as string) from $table") val (_, cometPlan) = checkSparkAnswerAndOperator(query) val project = stripAQEPlan(cometPlan).collectFirst { case p: CometProjectExec => p }.get - val id = project.expressions.head - CometSparkSessionExtensions.withFallbackReason(id, "reason 1") - CometSparkSessionExtensions.withFallbackReason(project, "reason 2") - CometSparkSessionExtensions.withFallbackReason(project, "reason 3", id) - CometSparkSessionExtensions.withFallbackReason(project, id) - CometSparkSessionExtensions.withFallbackReason(project, "reason 4") - CometSparkSessionExtensions.withFallbackReason(project, "reason 5", id) - CometSparkSessionExtensions.withFallbackReason(project, id) + // Reasons accumulate on the node they are recorded against, and are never overwritten. + // There is no roll-up here: a reason tagged on an expression is lifted onto the enclosing + // operator centrally by CometExecRule, not by withFallbackReason. + CometSparkSessionExtensions.withFallbackReason(project, "reason 1") + CometSparkSessionExtensions.withFallbackReason(project, "reason 2\nreason 3") + CometSparkSessionExtensions.withFallbackReasons(project, Set("reason 4", "reason 5")) CometSparkSessionExtensions.withFallbackReason(project, "reason 6") val explain = new ExtendedExplainInfo().generateExtendedInfo(project) for (i <- 1 until 7) { diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala index 8825b8af09..c71a4b7bed 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -33,7 +33,7 @@ import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAg import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ShuffleExchangeExec} import org.apache.spark.sql.types.{DataTypes, StructField, StructType} -import org.apache.comet.CometConf +import org.apache.comet.{CometConf, CometExplainInfo} import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator} @@ -77,6 +77,64 @@ class CometExecRuleSuite extends CometTestBase { }.sum } + test("expression-level fallback reasons are rolled up onto the operator that falls back") { + // Extended explain only walks plan nodes, so a reason recorded on a sub-expression is + // invisible unless CometExecRule lifts it onto the enclosing operator. Disabling a single + // expression makes the Project fall back with the reason living on the Multiply node. + // See https://github.com/apache/datafusion-comet/issues/5230. + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + + val sparkPlan = createSparkPlan(spark, "SELECT id * 2 as doubled FROM test_data") + assert(countOperators(sparkPlan, classOf[ProjectExec]) == 1) + + withSQLConf( + CometConf.getExprEnabledConfigKey("Multiply") -> "false", + CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val transformedPlan = applyCometExecRule(sparkPlan) + val project = stripAQEPlan(transformedPlan).collectFirst { case p: ProjectExec => p }.get + + val reasons = project + .getTagValue(CometExplainInfo.FALLBACK_REASONS) + .getOrElse(Set.empty[String]) + assert( + reasons.exists(_.contains("Multiply")), + s"expected the Multiply reason on the ProjectExec, got: $reasons") + // The generic catch-all message must not appear: a real reason was available. + assert( + !reasons.exists(_.contains("is not supported")), + s"a real reason was available but the generic message was used too: $reasons") + } + } + } + + test("strict fallback reason checking is off by default and on for Comet's own suites") { + // The strict check turns "a serde returned None without saying why" into a hard failure. It + // must stay off in production, where the generic " is not supported" message is the + // right user-facing behaviour, and on for every Comet suite so the bug class cannot ship + // again. Enabling it in CometTestBase is what actually exercises it: the whole test corpus + // runs with it on. See https://github.com/apache/datafusion-comet/issues/5230. + assert(!CometConf.COMET_STRICT_FALLBACK_REASONS.defaultValue.get) + assert(CometConf.COMET_STRICT_FALLBACK_REASONS.get(spark.sessionState.conf)) + } + + test("strict mode does not fire for operators Comet never attempted to convert") { + // Strict mode must only fire when a serde actually attempted the operator and declined. An + // operator Comet has no handler for was never attempted, so demanding a specific reason + // would be wrong - it keeps the generic message. + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + + val sparkPlan = createSparkPlan(spark, "SELECT id FROM test_data") + withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "false") { + // With local table scan disabled the leaf has no Comet handler applied, and planning + // must complete rather than throw. + val transformedPlan = applyCometExecRule(sparkPlan) + assert(transformedPlan != null) + } + } + } + test( "CometExecRule should apply basic operator transformations, but only when Comet is enabled") { withTempView("test_data") { diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala index b4442d5a23..f59ffa7d63 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala @@ -87,6 +87,10 @@ abstract class CometTestBase conf.set(CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.key, "true") conf.set(CometConf.COMET_ONHEAP_MEMORY_OVERHEAD.key, "2g") conf.set(CometConf.COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED.key, "true") + // Fail loudly if a serde declines an operator without stating why, rather than letting the + // generic " is not supported" message mask the missing reason. + // See https://github.com/apache/datafusion-comet/issues/5230. + conf.set(CometConf.COMET_STRICT_FALLBACK_REASONS.key, "true") // SortOrder is incompatible for mixed zero and negative zero floating point values, but // this is an edge case, and we expect most users to allow sorts on floating point, so we // enable this for the tests From 4abdc9842ff93b4dcad5b8f1d30e47db511020e0 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 3 Aug 2026 14:17:03 -0600 Subject: [PATCH 2/4] fix: remove unused val left by the roll-up removal in CometIn scalafix RemoveUnused flagged `val allExprs = list ++ Seq(value)`, which only existed to feed the deleted roll-up call. --- spark/src/main/scala/org/apache/comet/serde/predicates.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/predicates.scala b/spark/src/main/scala/org/apache/comet/serde/predicates.scala index 7ae1c35940..0e7bb02cd6 100644 --- a/spark/src/main/scala/org/apache/comet/serde/predicates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/predicates.scala @@ -385,7 +385,6 @@ object ComparisonUtils { .setIn(builder) .build()) } else { - val allExprs = list ++ Seq(value) None } } From bad0b27873a6e7d66d87d4722d23e2dcaf353179 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 3 Aug 2026 16:50:10 -0600 Subject: [PATCH 3/4] fix: lift fallback reasons off the rewritten tree in exprToProto Strict mode caught a real pre-existing hole. `exprToProto` runs `DecimalPrecision.promote`, and `transformUp` rebuilds every node on the path to a rewritten one, so for decimal arithmetic the nodes that serde actually converts are copies rather than the nodes in the plan. Any reason recorded during conversion landed on a copy, where neither extended explain nor the operator roll-up could ever see it. The old hand-written roll-up did not find these either - it read the original `projectList`, so the reason was equally lost - which is why this only surfaced now: previously the empty tag still rendered as a bare `[COMET: ]` and nobody noticed. TPC-DS q9's approved plan records exactly that, and is updated here to carry the real reason instead. Copy the reasons from the rewritten tree onto the original node, the same copy-back the `Invoke` / `StaticInvoke` rewrites in `Spark4xCometExprShim` already do. Repro (all Spark versions, [exec] and [expressions] CI shards): INSERT INTO t SELECT CAST(id AS decimal(18,4)) + 0.0001 FROM range(20000) threw "Comet did not convert Project but recorded no fallback reason". Also drops an empty `if (r.isEmpty) {}` block left in CometAlias by the roll-up removal. Verified: CometSqlFileTestSuite, the full [expressions] shard (1127 tests) and [exec] shard (508 tests) on Spark 4.0, and both TPC-DS plan stability suites (129 tests) on Spark 3.4, 3.5, 4.0 and 4.1 - all pass. Plan stability needs spark.test.home pointed at the Comet repo root to run. --- .../apache/comet/serde/QueryPlanSerde.scala | 22 ++++++++++++++++--- .../apache/comet/serde/namedExpressions.scala | 4 +--- .../approved-plans-v1_4/q9/extended.txt | 2 +- 3 files changed, 21 insertions(+), 7 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 87e69084f1..1bda4e7021 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -35,8 +35,8 @@ import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ -import org.apache.comet.CometConf -import org.apache.comet.CometSparkSessionExtensions.{withFallbackReason, withInfo} +import org.apache.comet.{CometConf, CometExplainInfo} +import org.apache.comet.CometSparkSessionExtensions.{withFallbackReason, withFallbackReasons, withInfo} import org.apache.comet.expressions._ import org.apache.comet.parquet.CometParquetUtils import org.apache.comet.serde.ExprOuterClass.{AggExpr, Expr, ScalarFunc} @@ -782,7 +782,23 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { binding: Boolean = true): Option[Expr] = { val newExpr = DecimalPrecision.promote(expr) - exprToProtoInternal(newExpr, inputs, binding) + val result = exprToProtoInternal(newExpr, inputs, binding) + if (result.isEmpty && !newExpr.eq(expr)) { + // `promote` rewrites decimal arithmetic, and `transformUp` rebuilds every node on the path + // to a rewritten one. Any fallback reason recorded while converting therefore landed on a + // copy that is not in the plan, where neither explain nor + // `CometExecRule.rollUpFallbackReasons` can see it. Lift the reasons onto the original node. + // Same copy-back the `Invoke` / `StaticInvoke` rewrites in `Spark4xCometExprShim` do. + val reasons = newExpr + .collect { case e: Expression => e } + .flatMap(_.getTagValue(CometExplainInfo.FALLBACK_REASONS)) + .flatten + .toSet + if (reasons.nonEmpty) { + withFallbackReasons(expr, reasons) + } + } + result } /** diff --git a/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala b/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala index 62c2e1eead..edd083c282 100644 --- a/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala +++ b/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala @@ -29,9 +29,7 @@ object CometAlias extends CometExpressionSerde[Alias] { a: Alias, inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { - val r = exprToProtoInternal(a.child, inputs, binding) - if (r.isEmpty) {} - r + exprToProtoInternal(a.child, inputs, binding) } } diff --git a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt index 53ba3252cf..daeac6e123 100644 --- a/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt +++ b/spark/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q9/extended.txt @@ -1,4 +1,4 @@ - Project [COMET: ] + Project [COMET: Unsupported data type: StructType(StructField(count(1),LongType,false),StructField(avg(ss_ext_discount_amt),DecimalType(11,6),true),StructField(avg(ss_net_paid),DecimalType(11,6),true))] : :- Subquery : : +- CometColumnarToRow : : +- CometProject From 49a4591b91397aff325f1674cca3f219eeaab5fe Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 4 Aug 2026 11:11:54 -0600 Subject: [PATCH 4/4] test: cover the strict fallback check directly, document its ordering dependency Addresses review feedback on #5236: - add a test that drives `reportUnexplainedFallback` with the exact shape a serde produces when it returns None without recording a reason (a handled operator over native children, no tag anywhere), asserting the strict-mode throw and the generic message when strict mode is off. No serde in the tree reaches that state - which is what the check enforces - so the operator is constructed by hand and the method is now package-visible. - document that `reportUnexplainedFallback` must run after `rollUpFallbackReasons`, since it reads only the operator's own tag. --- .../apache/comet/rules/CometExecRule.scala | 19 ++++++++- .../comet/rules/CometExecRuleSuite.scala | 40 +++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala index 9501cfba47..ef2f37371c 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala @@ -706,7 +706,10 @@ case class CometExecRule(session: SparkSession) if (converted.isEmpty) { // Comet looked at this operator and declined it, so it stays in the Spark plan. Lift any // reasons recorded on its expressions onto the operator itself - see - // `rollUpFallbackReasons` for why this is needed - and make sure something was recorded. + // `rollUpFallbackReasons` for why this is needed - and then make sure something was + // recorded. The order is required, not incidental: `reportUnexplainedFallback` inspects only + // the operator's own tag, so a reason still sitting on an expression would look like no + // reason at all and trip the strict check. rollUpFallbackReasons(op) reportUnexplainedFallback(op) } @@ -776,6 +779,8 @@ case class CometExecRule(session: SparkSession) * that actually failed conversion. That matters because some expression instances * (`AttributeReference`s, DPP subquery expressions) are shared across operators, so an unscoped * roll-up could surface one expression's reason under several unrelated operators. + * + * [[reportUnexplainedFallback]] relies on this having run first; the two must not be separated. */ private def rollUpFallbackReasons(op: SparkPlan): Unit = { val reasons = op.expressions @@ -797,8 +802,18 @@ case class CometExecRule(session: SparkSession) * own test suites) that is a hard failure; otherwise fall back to a generic message so users * still see something. The generic message is what used to mask this whole class of bug, which * is why the strict check exists. + * + * Must run *after* [[rollUpFallbackReasons]] for the same operator. The check reads only `op`'s + * own tag, because `hasFallbackReason` deliberately does not traverse expressions (it is a + * planning control signal, not explain output), so an expression-level reason that has not been + * lifted yet would be mistaken for no reason at all. [[convertToComet]] is the only production + * caller and keeps the two calls together. + * + * Package-visible so `CometExecRuleSuite` can drive the strict failure directly: no serde in + * the tree reaches this state, which is exactly what the check enforces, so the only way to + * test it is to construct the shape by hand. */ - private def reportUnexplainedFallback(op: SparkPlan): Unit = { + private[comet] def reportUnexplainedFallback(op: SparkPlan): Unit = { if (op.children.forall(_.isInstanceOf[CometNativeExec]) && !hasFallbackReason(op)) { if (CometConf.COMET_STRICT_FALLBACK_REASONS.get(op.conf)) { throw new IllegalStateException( diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala index c71a4b7bed..5444a89fa3 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala @@ -82,6 +82,8 @@ class CometExecRuleSuite extends CometTestBase { // invisible unless CometExecRule lifts it onto the enclosing operator. Disabling a single // expression makes the Project fall back with the reason living on the Multiply node. // See https://github.com/apache/datafusion-comet/issues/5230. + // This also pins the ordering inside `convertToComet`: strict mode is on in CometTestBase, so + // if the roll-up stopped running before the strict check, planning here would throw. withTempView("test_data") { createTestDataFrame.createOrReplaceTempView("test_data") @@ -108,6 +110,44 @@ class CometExecRuleSuite extends CometTestBase { } } + test("strict mode fails an operator that Comet declined without recording a reason") { + // The bug this guards against is a serde returning None and forgetting to say why, which the + // generic " is not supported" message used to hide. No serde in the tree is in that + // state (the whole test corpus runs with strict mode on, which is what enforces it), so drive + // the check directly with the shape such a serde produces: a handled operator whose children + // are all native and which carries no reason on itself or its expressions. + withTempView("test_data") { + createTestDataFrame.createOrReplaceTempView("test_data") + + val sparkPlan = createSparkPlan(spark, "SELECT id * 2 as doubled FROM test_data") + withSQLConf(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") { + val nativeChild = stripAQEPlan(applyCometExecRule(sparkPlan)).collectFirst { + case op: CometNativeExec => op + }.get + val rule = CometExecRule(spark) + + // ProjectExec has a registered serde, so Comet did attempt this operator. + val strictOp = ProjectExec(nativeChild.output, nativeChild) + assert(CometExecRule.allExecs.contains(strictOp.getClass)) + val e = intercept[IllegalStateException] { + rule.reportUnexplainedFallback(strictOp) + } + assert(e.getMessage.contains("recorded no fallback reason")) + assert(e.getMessage.contains(strictOp.nodeName)) + + // Production default: no throw, and the generic message so users still see something. + withSQLConf(CometConf.COMET_STRICT_FALLBACK_REASONS.key -> "false") { + val lenientOp = ProjectExec(nativeChild.output, nativeChild) + rule.reportUnexplainedFallback(lenientOp) + val reasons = lenientOp + .getTagValue(CometExplainInfo.FALLBACK_REASONS) + .getOrElse(Set.empty[String]) + assert(reasons == Set(s"${lenientOp.nodeName} is not supported")) + } + } + } + } + test("strict fallback reason checking is off by default and on for Comet's own suites") { // The strict check turns "a serde returned None without saying why" into a hard failure. It // must stay off in production, where the generic " is not supported" message is the