This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 5b8a45e061 refactor: replace hand-coded rollup of expression fallback
reasons onto operators (#5236)
5b8a45e061 is described below
commit 5b8a45e061691a076fefcf5911aa1dcdb8ed93e5
Author: Andy Grove <[email protected]>
AuthorDate: Tue Aug 4 13:42:36 2026 -0600
refactor: replace hand-coded rollup of expression fallback reasons onto
operators (#5236)
* 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 `"<operator> 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.
* 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.
* 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.
* 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.
---
.../main/scala/org/apache/comet/CometConf.scala | 14 +++
.../apache/comet/CometSparkSessionExtensions.scala | 62 +++++--------
.../org/apache/comet/expressions/CometCast.scala | 1 -
.../org/apache/comet/rules/CometExecRule.scala | 86 +++++++++++++++++-
.../comet/serde/CometBloomFilterMightContain.scala | 2 -
.../apache/comet/serde/CometScalarFunction.scala | 4 +-
.../org/apache/comet/serde/CometSortOrder.scala | 2 -
.../org/apache/comet/serde/QueryPlanSerde.scala | 59 ++++++------
.../scala/org/apache/comet/serde/aggregates.scala | 50 +++--------
.../scala/org/apache/comet/serde/arithmetic.scala | 8 +-
.../main/scala/org/apache/comet/serde/arrays.scala | 57 +++++-------
.../scala/org/apache/comet/serde/bitwise.scala | 4 +-
.../scala/org/apache/comet/serde/conditional.scala | 6 --
.../apache/comet/serde/contraintExpressions.scala | 8 +-
.../scala/org/apache/comet/serde/datetime.scala | 33 +++----
.../apache/comet/serde/decimalExpressions.scala | 6 +-
.../main/scala/org/apache/comet/serde/json.scala | 4 +-
.../main/scala/org/apache/comet/serde/maps.scala | 10 +--
.../main/scala/org/apache/comet/serde/math.scala | 26 +++---
.../org/apache/comet/serde/namedExpressions.scala | 6 +-
.../scala/org/apache/comet/serde/predicates.scala | 6 +-
.../scala/org/apache/comet/serde/statics.scala | 11 ++-
.../scala/org/apache/comet/serde/strings.scala | 20 ++---
.../scala/org/apache/comet/serde/structs.scala | 5 +-
.../scala/org/apache/comet/serde/unixtime.scala | 5 +-
.../main/scala/org/apache/comet/serde/url.scala | 4 +-
.../apache/spark/sql/comet/CometWindowExec.scala | 32 +++----
.../org/apache/spark/sql/comet/operators.scala | 34 ++-----
.../apache/comet/serde/CometToPrettyString.scala | 2 -
.../org/apache/comet/shims/CometExprShim.scala | 8 +-
.../org/apache/comet/serde/CometMapSort.scala | 4 +-
.../apache/comet/serde/CometToPrettyString.scala | 2 -
.../org/apache/comet/shims/CometExprShim4x.scala | 8 +-
.../approved-plans-v1_4/q9/extended.txt | 2 +-
.../org/apache/comet/CometExpressionSuite.scala | 14 ++-
.../apache/comet/rules/CometExecRuleSuite.scala | 100 ++++++++++++++++++++-
.../scala/org/apache/spark/sql/CometTestBase.scala | 4 +
37 files changed, 392 insertions(+), 317 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala
b/spark/src/main/scala/org/apache/comet/CometConf.scala
index e9cf8ca010..1bb524349d 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 " +
+ "'<operator> 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 ca4e1e4e79..3bcb3f98e1 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 a709129bac..ef2f37371c 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,23 @@ 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 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)
+ }
+ 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 +762,69 @@ 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.
+ *
+ * [[reportUnexplainedFallback]] relies on this having run first; the two
must not be separated.
+ */
+ 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.
+ *
+ * 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[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(
+ 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 9e867aad2d..cad5da6746 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.types._
import org.apache.comet.CometConf
import org.apache.comet.CometExplainInfo
-import org.apache.comet.CometSparkSessionExtensions.{appendTagValues,
withFallbackReason, withInfo, withNativeExpr}
+import org.apache.comet.CometSparkSessionExtensions.{appendTagValues,
withFallbackReason, withFallbackReasons, withInfo, withNativeExpr}
import org.apache.comet.expressions._
import org.apache.comet.parquet.CometParquetUtils
import org.apache.comet.serde.ExprOuterClass.{AggExpr, Expr, ScalarFunc}
@@ -725,10 +725,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
}
@@ -745,7 +742,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)
@@ -793,11 +789,14 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
val newExpr = DecimalPrecision.promote(expr)
val result = exprToProtoInternal(newExpr, inputs, binding)
if (!(newExpr eq expr)) {
- // `promote` rebuilt the tree, so the coverage tags landed on copies
that the operator does
- // not hold. Lift them onto `expr` so
`CometExecRule.rollUpInfoMessages`, which walks the
- // operator's own expressions, still sees them. Skipped in the common
case where `promote`
- // returned the same tree and there is nothing to lift.
+ // `promote` rebuilt the tree, so the tags landed on copies that the
operator does not hold.
+ // Lift them onto `expr` so the roll-ups in `CometExecRule`, which walk
the operator's own
+ // expressions, still see them. Skipped in the common case where
`promote` returned the same
+ // tree and there is nothing to lift.
liftCoverageTags(newExpr, expr)
+ if (result.isEmpty) {
+ liftFallbackReasons(newExpr, expr)
+ }
}
result
}
@@ -813,6 +812,27 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
appendTagValues(to, CometExplainInfo.CODEGEN_DISPATCH_EXPRS,
dispatched.toSet)
}
+ /**
+ * Lift fallback reasons recorded anywhere in the rewritten tree onto the
original node, so that
+ * `CometExecRule.rollUpFallbackReasons` and extended explain can see them.
Without this the
+ * reason is attached to a copy that is not in the plan and is lost entirely
- see
+ * https://github.com/apache/datafusion-comet/issues/5230. Same copy-back
that the `Invoke` /
+ * `StaticInvoke` rewrites in `Spark4xCometExprShim` do.
+ *
+ * Only called when conversion failed: a fallback reason states why an
expression could not be
+ * converted, so lifting one off a tree that converted fine would attribute
a stale reason to an
+ * operator that has no problem.
+ */
+ private def liftFallbackReasons(from: Expression, to: Expression): Unit = {
+ val reasons = mutable.Set.empty[String]
+ from.foreach { e =>
+ e.getTagValue(CometExplainInfo.FALLBACK_REASONS).foreach(reasons ++= _)
+ }
+ if (reasons.nonEmpty) {
+ withFallbackReasons(to, reasons.toSet)
+ }
+ }
+
/**
* Convert a Spark expression to a protocol-buffer representation of a
native Comet/DataFusion
* expression.
@@ -917,7 +937,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
}
})
@@ -986,7 +1006,6 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
.newBuilder(),
inner).build())
} else {
- withFallbackReason(expr, child)
None
}
}
@@ -1016,7 +1035,6 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
.newBuilder(),
inner).build())
} else {
- withFallbackReason(expr, left, right)
None
}
}
@@ -1045,7 +1063,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
@@ -1128,20 +1145,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 8023231b35..058ff2ea5f 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)
@@ -426,7 +423,7 @@ object CometRound extends CometExpressionSerde[Round] {
r.ansiEnabled,
childExpr,
scaleExpr)
- optExprWithFallbackReason(optExpr, r, r.child)
+ optExpr
}
}
@@ -451,7 +448,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 f41e0fafaa..b2bfe6ee06 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
}
}
@@ -58,7 +58,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..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,11 +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) {
- withFallbackReason(a, a.child)
- }
- r
+ exprToProtoInternal(a.child, inputs, binding)
}
}
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..0e7bb02cd6 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
}
}
@@ -385,8 +385,6 @@ object ComparisonUtils {
.setIn(builder)
.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 05439a16e3..74ef3af91c 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
@@ -1056,7 +1056,6 @@ object CometProjectExec extends
CometOperatorSerde[ProjectExec] {
.addAllProjectList(exprs.map(_.get).asJava)
Some(builder.setProjection(projectBuilder).build())
} else {
- withFallbackReason(op, op.projectList: _*)
None
}
}
@@ -1116,7 +1115,6 @@ object CometFilterExec extends
CometOperatorSerde[FilterExec] {
.setPredicate(cond.get)
Some(builder.setFilter(filterBuilder).build())
} else {
- withFallbackReason(op, op.condition, op.child)
None
}
}
@@ -1189,7 +1187,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
}
}
@@ -1379,11 +1377,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
@@ -1392,7 +1386,6 @@ object CometExpandExec extends
CometOperatorSerde[ExpandExec] {
.setNumExprPerProject(op.projections.head.size)
Some(builder.setExpand(expandBuilder).build())
} else {
- withFallbackReason(op, allProjExprs: _*)
None
}
}
@@ -1472,7 +1465,6 @@ object CometExplodeExec extends
CometOperatorSerde[GenerateExec] {
val childExprProto = exprToProto(childExpr, op.child.output)
if (childExprProto.isEmpty) {
- withFallbackReason(op, childExpr)
return None
}
@@ -1484,7 +1476,6 @@ object CometExplodeExec extends
CometOperatorSerde[GenerateExec] {
}
if (projectExprs.exists(_.isEmpty) || childOp.isEmpty) {
- withFallbackReason(op, op.output: _*)
return None
}
@@ -1804,10 +1795,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
}
@@ -1848,9 +1836,6 @@ trait CometBaseAggregate {
Some(builder.setHashAgg(hashAggBuilder).build())
}
} else {
- val allChildren: Seq[Expression] =
- groupingExpressions ++ aggregateExpressions ++ aggregateAttributes
- withFallbackReason(aggregate, allChildren: _*)
None
}
}
@@ -1879,8 +1864,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
@@ -2163,7 +2147,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
@@ -2202,8 +2185,6 @@ trait CometHashJoin {
condition.foreach(joinBuilder.setCondition)
Some(builder.setHashJoin(joinBuilder).build())
} else {
- val allExprs: Seq[Expression] = joinKeys
- withFallbackReason(join, allExprs: _*)
None
}
}
@@ -2330,7 +2311,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
@@ -2648,15 +2628,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
@@ -2717,8 +2695,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/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 762d754ec4..a6278c7383 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
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..5444a89fa3 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,104 @@ 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.
+ // 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")
+
+ 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 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 "<operator> 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 "<operator> 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 "<operator> 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
