You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/23 10:40:28 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-7095-6cc7b8625aa9f33c24de221182c59bc7fb58c97e created (now 886e0a9532)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7095-6cc7b8625aa9f33c24de221182c59bc7fb58c97e
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 886e0a9532 fix: scope fallback reasons to the current conversion (#7095)
This branch includes the following new commits:
new 886e0a9532 fix: scope fallback reasons to the current conversion (#7095)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: scope fallback reasons to the current conversion (#7095)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7095-6cc7b8625aa9f33c24de221182c59bc7fb58c97e
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 886e0a9532597322fc131027f4d1841f1e36ba66
Author: Yu Hsun Hung <75...@users.noreply.github.com>
AuthorDate: Fri Oct 23 09:38:28 2026 +0000
fix: scope fallback reasons to the current conversion (#7095)
* fix: scope fallback reasons to the current conversion
* fix: preserve aggregate preflight fallback reasons
* fix: scope range shuffle fallback reasons and simplify expression journal
---
.../apache/comet/CometSparkSessionExtensions.scala | 22 +-
.../org/apache/comet/ExtendedExplainInfo.scala | 37 ++++
.../org/apache/comet/rules/CometExecRule.scala | 55 +++--
.../org/apache/comet/serde/QueryPlanSerde.scala | 78 ++++---
.../shuffle/CometShuffleExchangeExec.scala | 10 +-
.../apache/comet/shims/Spark4xCometExprShim.scala | 15 +-
.../comet/exec/CometNativeShuffleSuite.scala | 52 ++++-
.../apache/comet/exec/CometWindowExecSuite.scala | 52 ++++-
.../apache/comet/rules/CometExecRuleSuite.scala | 238 ++++++++++++++++++++-
.../sql/comet/CometDecimalPromotionSuite.scala | 34 ++-
10 files changed, 515 insertions(+), 78 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
index 11472de837..b9880d1d03 100644
--- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
+++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
@@ -24,6 +24,7 @@ import java.nio.ByteOrder
import org.apache.spark.{SparkConf, SparkEnv}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{SparkSession, SparkSessionExtensions}
+import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.{TreeNode, TreeNodeTag}
import org.apache.spark.sql.comet._
@@ -317,9 +318,12 @@ object CometSparkSessionExtensions extends Logging {
* 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,
- * each new reason is also emitted as a warning.
+ * Reasons are accumulated on the node's `FALLBACK_REASONS` tag and are surfaced in extended
+ * explain output. During conversion, expression reasons accumulate only within that scope:
+ * pre-existing expression tags may have been copied from an unrelated conversion. Operator
+ * reasons remain sticky because earlier planning rules can intentionally veto conversion. When
+ * `COMET_EXPLAIN_FALLBACK_LOG_ENABLED` is enabled, each new reason is also emitted as a
+ * warning.
*
* @param node
* The Spark operator or expression that is falling back to Spark.
@@ -336,9 +340,15 @@ object CometSparkSessionExtensions extends Logging {
logWarning(s"Comet cannot accelerate ${node.getClass.getSimpleName} because: $reason")
}
}
- val existingNodeInfos =
- node.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String])
- node.setTagValue(CometExplainInfo.FALLBACK_REASONS, existingNodeInfos ++ info)
+ val reasons = node match {
+ case expr: Expression =>
+ CometExplainInfo.recordFallbackReasons(expr, info).getOrElse {
+ expr.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String]) ++ info
+ }
+ case _ =>
+ node.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String]) ++ info
+ }
+ node.setTagValue(CometExplainInfo.FALLBACK_REASONS, reasons)
node
}
diff --git a/spark/src/main/scala/org/apache/comet/ExtendedExplainInfo.scala b/spark/src/main/scala/org/apache/comet/ExtendedExplainInfo.scala
index 07a7f57ccd..8638787f90 100644
--- a/spark/src/main/scala/org/apache/comet/ExtendedExplainInfo.scala
+++ b/spark/src/main/scala/org/apache/comet/ExtendedExplainInfo.scala
@@ -317,6 +317,43 @@ object CometCoverageStats {
object CometExplainInfo {
val FALLBACK_REASONS = new TreeNodeTag[Set[String]]("CometFallbackReasons")
+
+ // Planning is synchronous on the calling thread. Keep the journal scoped to a conversion,
+ // rather than trusting mutable tags which Catalyst can copy onto shared expression instances.
+ // Identity keys distinguish equal expressions. This is deliberately not inheritable: planning
+ // a subquery on another thread must not retain or share its parent's mutable journal.
+ private val fallbackJournal = new ThreadLocal[IdentityHashMap[Expression, Set[String]]]
+
+ private[comet] def withFallbackReasonScope[T](body: => T): T = {
+ val previous = fallbackJournal.get()
+ fallbackJournal.set(new IdentityHashMap[Expression, Set[String]]())
+ try body
+ finally {
+ if (previous == null) fallbackJournal.remove() else fallbackJournal.set(previous)
+ }
+ }
+
+ private[comet] def withExpressionFallbackReasonScope[T](body: => T): T = {
+ if (fallbackJournal.get() != null) body else withFallbackReasonScope(body)
+ }
+
+ private[comet] def recordFallbackReasons(
+ node: Expression,
+ reasons: Set[String]): Option[Set[String]] = {
+ Option(fallbackJournal.get()).map { journal =>
+ val previous = Option(journal.get(node)).getOrElse(Set.empty[String])
+ val current = previous ++ reasons
+ journal.put(node, current)
+ current
+ }
+ }
+
+ private[comet] def currentFallbackReasons(node: Expression): Set[String] = {
+ Option(fallbackJournal.get())
+ .flatMap(journal => Option(journal.get(node)))
+ .getOrElse(Set.empty)
+ }
+
val EXTENSION_INFO = new TreeNodeTag[Set[String]]("CometExtensionInfo")
// Expression names the serde routed through the JVM codegen dispatcher. Set on each such
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 02e4cff1a4..87e5aac0fc 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
@@ -1115,20 +1115,21 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
}
/** 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)
+ private def convertToComet(op: SparkPlan, handler: CometOperatorSerde[_]): Option[SparkPlan] =
+ CometExplainInfo.withFallbackReasonScope {
+ 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. `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. Keep these two calls together and in this order.
+ rollUpFallbackReasons(op)
+ reportUnexplainedFallback(op)
+ }
+ converted
}
- converted
- }
private def tryConvertToComet(
op: SparkPlan,
@@ -1196,8 +1197,10 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
* 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`.
+ * Only reasons recorded for these expression instances during this conversion are collected.
+ * Tags alone are insufficient: Catalyst may have copied one onto a shared singleton. 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
@@ -1206,11 +1209,10 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
*
* [[reportUnexplainedFallback]] relies on this having run first; the two must not be separated.
*/
- private def rollUpFallbackReasons(op: SparkPlan): Unit = {
+ private[comet] def rollUpFallbackReasons(op: SparkPlan): Unit = {
val reasons = op.expressions
.flatMap(_.collect { case e: Expression => e })
- .flatMap(_.getTagValue(CometExplainInfo.FALLBACK_REASONS))
- .flatten
+ .flatMap(CometExplainInfo.currentFallbackReasons)
.toSet
if (reasons.nonEmpty) {
val _ = withFallbackReasons(op, reasons)
@@ -1941,7 +1943,22 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
* this tagging pass will drift and either crash (missed tag) or over-disable (spurious tag). A
* shared predicate helper would be preferable.
*/
- private def canAggregateBeConverted(
+ private[rules] def canAggregateBeConverted(
+ agg: BaseAggregateExec,
+ expectedMode: AggregateMode): Boolean =
+ CometExplainInfo.withFallbackReasonScope {
+ val convertible = checkAggregateConversion(agg, expectedMode)
+ if (!convertible) {
+ // This preflight may discover an expression failure that actual conversion never reaches:
+ // the child can stay in Spark, making conversion decline on buffer compatibility first.
+ // Preserve this probe's diagnostics on their operator before discarding its journal.
+ // Reading only the journal keeps copied or stale expression tags out of the explanation.
+ rollUpFallbackReasons(agg)
+ }
+ convertible
+ }
+
+ private def checkAggregateConversion(
agg: BaseAggregateExec,
expectedMode: AggregateMode): Boolean = {
val handler = allExecs.get(agg.getClass)
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 afa51ee6ab..7ee6b85fd0 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -936,22 +936,44 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
def exprToProto(
expr: Expression,
inputs: Seq[Attribute],
- binding: Boolean = true): Option[Expr] = {
-
- val newExpr = DecimalPrecision.promote(expr)
- val result = exprToProtoInternal(newExpr, inputs, binding)
- if (!(newExpr eq expr)) {
- // `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)
+ binding: Boolean = true): Option[Expr] =
+ CometExplainInfo.withExpressionFallbackReasonScope {
+ val newExpr = DecimalPrecision.promote(expr)
+ val result = exprToProtoInternal(newExpr, inputs, binding)
+ if (!(newExpr eq expr)) {
+ // `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
+ }
+
+ /**
+ * Convert an independent expression and return its current failure reasons before the scope
+ * closes. Used by planning probes outside operator conversion, such as native range shuffle.
+ * Ignore mutable tags and keep the probe's journal separate from any enclosing conversion.
+ */
+ def exprToProtoWithFallbackReasons(
+ expr: Expression,
+ inputs: Seq[Attribute],
+ binding: Boolean = true): (Option[Expr], Set[String]) =
+ CometExplainInfo.withFallbackReasonScope {
+ val result = exprToProto(expr, inputs, binding)
+ val reasons = if (result.isEmpty) {
+ expr
+ .collect { case e: Expression => e }
+ .flatMap(CometExplainInfo.currentFallbackReasons)
+ .toSet
+ } else {
+ Set.empty[String]
+ }
+ (result, reasons)
}
- result
- }
private def liftCoverageTags(from: Expression, to: Expression): Unit = {
val exprs = from.collect { case e: Expression => e }
@@ -961,16 +983,19 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
}
/**
- * 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.
+ * Lift fallback reasons recorded during the current conversion 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.
*
* Callers are the serde paths that rebuild an expression tree before converting it, so the
* reasons land on copies the operator does not hold: `DecimalPrecision.promote` here, and
* `CometWindowExec`'s `DecimalAggregates` unwrapping.
*
+ * Reads the conversion journal, not tags: a shared literal may carry a tag copied from an
+ * unrelated expression or an earlier query. Callers must run inside a fallback reason scope.
+ *
* 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.
@@ -978,7 +1003,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
def liftFallbackReasons(from: Expression, to: Expression): Unit = {
val reasons = mutable.Set.empty[String]
from.foreach { e =>
- e.getTagValue(CometExplainInfo.FALLBACK_REASONS).foreach(reasons ++= _)
+ reasons ++= CometExplainInfo.currentFallbackReasons(e)
}
if (reasons.nonEmpty) {
val _ = withFallbackReasons(to, reasons.toSet)
@@ -992,7 +1017,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
* The caller owns decimal promotion: this method serializes children of an already-promoted
* root without traversing them again. Literals and wrappers that introduce no decimal
* arithmetic can also use this path. Newly synthesized arithmetic must enter through
- * [[exprToProto]].
+ * [[exprToProto]]. Callers must run inside a fallback reason scope owned by the outer
+ * [[exprToProto]] call or the enclosing operator conversion.
*
* @param expr
* The input expression
@@ -1009,7 +1035,6 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
expr: Expression,
inputs: Seq[Attribute],
binding: Boolean): Option[Expr] = {
-
def convert[T <: Expression](expr: T, handler: CometExpressionSerde[T]): Option[Expr] = {
val exprConfName = handler.getExprConfigName(expr)
if (!CometConf.isExprEnabled(exprConfName)) {
@@ -1050,9 +1075,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
} else {
// Expressions that opt in via `CodegenDispatchFallback` route their `Incompatible`
// result through the JVM codegen dispatcher (Spark's own `doGenCode` inside the Comet
- // pipeline) so the projection stays native while still matching Spark. Everything else
- // falls back to Spark. Falling back is also the result when the dispatcher cannot
- // handle the expression.
+ // pipeline) so the projection stays native while still matching Spark. Everything
+ // else falls back to Spark, as does a dispatcher that cannot handle the expression.
val dispatched = dispatchIfFallback(handler, expr, inputs, binding).map {
case (h, proto) =>
val key = h.nativeOptInConfigKeyOverride
@@ -1107,8 +1131,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
expr.getTagValue(CometExplainInfo.DISPATCHED_SELF).isEmpty) {
withNativeExpr(expr, CometExplainInfo.exprDisplayName(expr))
}
- // Passthrough serdes such as Alias return an already-identified child expression. Preserve
- // that child's context instead of replacing it with the structural wrapper's origin.
+ // Passthrough serdes such as Alias return an already-identified child expression.
+ // Preserve that child's context instead of replacing it with the wrapper's origin.
if (protoExpr.hasExprId) protoExpr else attachExprIdAndContext(expr, protoExpr)
}
}
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
index f67ebb776e..616a9716c8 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
@@ -51,7 +51,7 @@ import org.apache.spark.util.random.XORShiftRandom
import com.google.common.base.Objects
-import org.apache.comet.{CometConf, CometExplainInfo}
+import org.apache.comet.CometConf
import org.apache.comet.CometConf.{COMET_SHUFFLE_ENABLED, COMET_SHUFFLE_MODE}
import org.apache.comet.CometSparkSessionExtensions.{cometCelebornShuffleFallbackReason, hasFallbackReason, isCometCelebornShuffleManagerEnabled, isCometShuffleManagerEnabled, isSpark40Plus, withFallbackReasons}
import org.apache.comet.serde.{Compatible, OperatorOuterClass, QueryPlanSerde, SupportLevel, Unsupported}
@@ -717,11 +717,11 @@ object CometShuffleExchangeExec
return reasons.toSeq
}
for (o <- orderings) {
- if (QueryPlanSerde.exprToProto(o, inputs).isEmpty) {
+ val (result, expressionReasons) =
+ QueryPlanSerde.exprToProtoWithFallbackReasons(o, inputs)
+ if (result.isEmpty) {
reasons += s"unsupported range partitioning sort order: $o"
- // Roll up fallback reasons recorded on the sort-order expression (e.g. strict
- // floating-point sort) so they surface in the shuffle's explain output.
- o.getTagValue(CometExplainInfo.FALLBACK_REASONS).foreach(reasons ++= _)
+ reasons ++= expressionReasons
}
}
for (dt <- orderings.map(_.dataType).distinct) {
diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
index de35673168..87f2731d97 100644
--- a/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
@@ -25,11 +25,10 @@ import org.apache.spark.sql.catalyst.expressions.json.{JsonExpressionUtils, Stru
import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.expressions.url.ParseUrlEvaluator
-import org.apache.comet.CometExplainInfo
import org.apache.comet.expressions.CometEvalMode
import org.apache.comet.serde.{CometAggregateExpressionSerde, CometExpressionSerde, CometHllSketchAgg, CometHllSketchEstimate, CometHllUnion, CometHllUnionAgg, CometListAgg, CometMapSort, CometRandStr, CometToPrettyString}
import org.apache.comet.serde.ExprOuterClass.Expr
-import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal
+import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, liftFallbackReasons}
/**
* Shared trait body for the Spark 4.x `CometExprShim` traits (4.0/4.1/4.2). Holds the parts that
@@ -86,18 +85,14 @@ trait Spark4xCometExprShim extends CometExprShim4x {
val toJson = StructsToJson(evaluator.options, child, evaluator.timeZoneId)
val exprProto = exprToProtoInternal(toJson, inputs, binding)
if (exprProto.isEmpty) {
- toJson
- .getTagValue(CometExplainInfo.FALLBACK_REASONS)
- .foreach(reasons => i.setTagValue(CometExplainInfo.FALLBACK_REASONS, reasons))
+ liftFallbackReasons(toJson, i)
}
exprProto
case (Literal(evaluator: ParseUrlEvaluator, _), "evaluate", args) =>
val parseUrl = ParseUrl(args, evaluator.failOnError)
val result = exprToProtoInternal(parseUrl, inputs, binding)
if (result.isEmpty) {
- parseUrl
- .getTagValue(CometExplainInfo.FALLBACK_REASONS)
- .foreach(reasons => i.setTagValue(CometExplainInfo.FALLBACK_REASONS, reasons))
+ liftFallbackReasons(parseUrl, i)
}
result
case _ => None
@@ -109,9 +104,7 @@ trait Spark4xCometExprShim extends CometExprShim4x {
val lengthOfJsonArray = LengthOfJsonArray(child)
val exprProto = exprToProtoInternal(lengthOfJsonArray, inputs, binding)
if (exprProto.isEmpty) {
- lengthOfJsonArray
- .getTagValue(CometExplainInfo.FALLBACK_REASONS)
- .foreach(reasons => s.setTagValue(CometExplainInfo.FALLBACK_REASONS, reasons))
+ liftFallbackReasons(lengthOfJsonArray, s)
}
exprProto
case _ => None
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
index 1135075914..06660cef2a 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
@@ -37,10 +37,10 @@ import org.apache.arrow.vector.types.pojo.{Field, Schema}
import org.apache.hadoop.fs.Path
import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.sql.{CometTestBase, DataFrame, Dataset, Row}
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.{Ascending, AttributeReference, SortOrder}
import org.apache.spark.sql.catalyst.expressions.aggregate.Final
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
-import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, RangePartitioning}
import org.apache.spark.sql.comet.{CometExec, CometHashAggregateExec, CometLocalTableScanExec, CometMetricNode, CometNativeExec, CometNativeScanExec, CometScanWrapper, CometSparkToColumnarExec, CometTakeOrderedAndProjectExec}
import org.apache.spark.sql.comet.execution.arrow.{CometArrowStream, CometNativeArrowSource}
import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec}
@@ -54,7 +54,7 @@ import org.apache.spark.sql.types.{ArrayType, DataType, IntegerType, LongType, M
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.comet.{CometConf, CometExecIterator, CometExplainInfo, CometShuffleBlockIterator, CometShuffleSizeLimitException, Native}
-import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus}
+import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, withFallbackReason}
import org.apache.comet.serde.{OperatorOuterClass, PartitioningOuterClass}
import org.apache.comet.shuffle.ShufflePartitionPusher
@@ -474,6 +474,52 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper
}
}
+ test("range shuffle reports current child failure reasons instead of stale sort-order tags") {
+ withSQLConf(CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "true") {
+ val attr = AttributeReference("key", LongType)()
+ val leaf = spark.sessionState.planner
+ .plan(LocalRelation(Seq(attr)))
+ .next()
+ .asInstanceOf[LocalTableScanExec]
+ val child = CometScanWrapper(OperatorOuterClass.Operator.getDefaultInstance, leaf)
+ val ordering = SortOrder(attr, Ascending)
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ val stale = "stale sort-order reason from an earlier conversion"
+ val key = CometConf.getExprEnabledConfigKey("AttributeReference")
+ val expected = s"Expression support is disabled. Set $key=true to enable it."
+ ordering.setTagValue(tag, Set(stale))
+
+ def check(): Unit = {
+ val exchange = ShuffleExchangeExec(RangePartitioning(Seq(ordering), 2), child)
+ assert(CometShuffleExchangeExec.shuffleSupported(exchange).isEmpty)
+ val reasons = exchange.getTagValue(tag).getOrElse(Set.empty[String])
+ assert(reasons.contains(expected), reasons.toString)
+ assert(!reasons.contains(stale), reasons.toString)
+ // The failure was on the child, so the root's copied tag really is still present.
+ assert(ordering.getTagValue(tag).contains(Set(stale)))
+ assert(
+ new org.apache.comet.ExtendedExplainInfo()
+ .generateExtendedInfo(exchange)
+ .contains(expected))
+ }
+
+ withSQLConf(key -> "false") {
+ check()
+ CometExplainInfo.withFallbackReasonScope {
+ // A probe must neither consume nor overwrite an enclosing conversion's journal.
+ withFallbackReason(attr, "outer conversion")
+ check()
+ assert(CometExplainInfo.currentFallbackReasons(attr) == Set("outer conversion"))
+ }
+ }
+ withSQLConf(key -> "true") {
+ val exchange = ShuffleExchangeExec(RangePartitioning(Seq(ordering), 2), child)
+ assert(CometShuffleExchangeExec.shuffleSupported(exchange).contains(CometNativeShuffle))
+ assert(exchange.getTagValue(tag).isEmpty)
+ }
+ }
+ }
+
test("wide decimals remain supported in shuffle payloads, ranges and single partitions") {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
index c0c3ccfec5..642a73e337 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
@@ -27,15 +27,15 @@ import org.scalatest.Tag
import org.apache.hadoop.fs.Path
import org.apache.spark.sql.{CometTestBase, DataFrame, Row}
import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, Divide, Expression, MakeDecimal, WindowExpression}
-import org.apache.spark.sql.comet.{CometSortExec, CometWindowExec, CometWindowGroupLimitExec}
-import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.comet.{CometColumnarToRowExec, CometNativeExec, CometSortExec, CometWindowExec, CometWindowGroupLimitExec}
+import org.apache.spark.sql.execution.{InputAdapter, SparkPlan, WholeStageCodegenExec}
import org.apache.spark.sql.execution.window.{WindowExec => SparkWindowExec}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{count, expr, lead, sum}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.DecimalType
-import org.apache.comet.CometConf
+import org.apache.comet.{CometConf, CometExplainInfo, ExtendedExplainInfo}
import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus}
class CometWindowExecSuite extends CometTestBase {
@@ -803,6 +803,52 @@ class CometWindowExecSuite extends CometTestBase {
}
}
+ Seq("Sum", "Average").foreach { name =>
+ test(s"issue #5499: rewritten decimal window $name retains fallback reasons") {
+ withTempView("decimal_window_fallback") {
+ Seq((1, "10.10"), (1, "20.25"), (2, "30.33"))
+ .toDF("g", "raw_v")
+ .selectExpr("g", "CAST(raw_v AS DECIMAL(8,2)) AS v")
+ .createOrReplaceTempView("decimal_window_fallback")
+ val key = CometConf.getExprEnabledConfigKey(name)
+ withSQLConf(
+ key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "false",
+ CometConf.COMET_STRICT_FALLBACK_REASONS.key -> "true") {
+ val function = if (name == "Sum") "SUM" else "AVG"
+ val df = sql(s"""SELECT $function(v) OVER (
+ PARTITION BY g ORDER BY v
+ ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
+ FROM decimal_window_fallback""")
+ val plan = df.queryExecution.executedPlan
+ if (name == "Sum") assertSparkPlanHasDecimalSumRewrite(plan)
+ else if (isSpark40Plus) assertSparkPlanHasDecimalAvgRewrite(plan)
+ val window = collect(plan) { case w: SparkWindowExec => w }.head
+ // Spark inserts the row transition after Comet conversion. The strict fallback check
+ // saw this native child before that transition was added.
+ def unwrapTransitions(child: SparkPlan): SparkPlan = child match {
+ case stage: WholeStageCodegenExec => unwrapTransitions(stage.child)
+ case adapter: InputAdapter => unwrapTransitions(adapter.child)
+ case transition: CometColumnarToRowExec => unwrapTransitions(transition.child)
+ case other => other
+ }
+ assert(unwrapTransitions(window.child).isInstanceOf[CometNativeExec], plan.toString)
+ val expected = Set(s"Expression support is disabled. Set $key=true to enable it.")
+ assert(window.getTagValue(CometExplainInfo.FALLBACK_REASONS).contains(expected))
+ // The original owner must carry the reason from the reconstructed window expression.
+ if (name == "Sum" || isSpark40Plus) {
+ assert(
+ window.windowExpression.head
+ .getTagValue(CometExplainInfo.FALLBACK_REASONS)
+ .contains(expected))
+ }
+ assert(new ExtendedExplainInfo().generateExtendedInfo(plan).contains(expected.head))
+ }
+ }
+ }
+ }
+
test("window: decimal AVG with PARTITION BY and ORDER BY") {
withTempDir { dir =>
Seq((1, "10.10"), (1, "20.25"), (1, "30.33"), (1, "41.00"), (2, "11.11"), (2, "22.22"))
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 62d968c95d..1ae3ab179c 100644
--- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala
@@ -19,7 +19,7 @@
package org.apache.comet.rules
-import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.{ConcurrentLinkedQueue, FutureTask, TimeUnit}
import scala.jdk.CollectionConverters._
import scala.util.Random
@@ -29,8 +29,9 @@ import org.scalatest.PrivateMethodTester._
import org.apache.logging.log4j.Level
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.FunctionIdentifier
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, ExpressionInfo, In, InSet, KnownFloatingPointNormalized, Literal, Not}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Expression, ExpressionInfo, In, InSet, KnownFloatingPointNormalized, Literal, Not, RuntimeReplaceable}
import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate, Final, Min, Partial, PartialMerge}
+import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero
import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, RoundRobinPartitioning}
@@ -873,6 +874,239 @@ class CometExecRuleSuite extends CometTestBase {
}
}
+ if (isSpark40Plus) {
+ Seq(
+ ("StructsToJson", "to_json(named_struct('value', id))"),
+ ("ParseUrl", "parse_url(concat('https://example.com/', id), 'HOST')"),
+ ("LengthOfJsonArray", "json_array_length(concat('[', id, ']'))")).foreach {
+ case (name, sqlExpression) =>
+ test(s"issue #5499: $name shim retains current fallback reasons on its wrapper") {
+ val plan = sql(s"SELECT $sqlExpression FROM range(2)").queryExecution.analyzed
+ val original = plan.expressions
+ .flatMap(_.collect {
+ case expr: RuntimeReplaceable if expr.getClass.getSimpleName == name => expr
+ })
+ .head
+ val wrapper = original.replacement
+ if (name == "LengthOfJsonArray") assert(wrapper.isInstanceOf[StaticInvoke])
+ else assert(wrapper.isInstanceOf[Invoke])
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ wrapper.setTagValue(tag, Set("unrelated conversion"))
+ val key = CometConf.getExprEnabledConfigKey(name)
+ val expected = Set(s"Expression support is disabled. Set $key=true to enable it.")
+ withSQLConf(key -> "false") {
+ CometExplainInfo.withFallbackReasonScope {
+ // Call the shim directly so the generic Invoke handler cannot supply an unrelated
+ // reason and make this test pass when copy-back is broken.
+ assert(
+ QueryPlanSerde
+ .sparkVersionSpecificExprToProtoInternal(
+ wrapper,
+ plan.children.head.output,
+ binding = true)
+ .isEmpty)
+ assert(wrapper.getTagValue(tag).contains(expected))
+ assert(CometExplainInfo.currentFallbackReasons(wrapper) == expected)
+ }
+ }
+ }
+ }
+ }
+
+ test("fallback reasons accumulate on unscoped expressions and remain sticky on operators") {
+ val expr = Literal(42)
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ withFallbackReason(expr, "unscoped first")
+ withFallbackReason(expr, "unscoped second")
+ assert(expr.getTagValue(tag).contains(Set("unscoped first", "unscoped second")))
+
+ val op = spark.sessionState.planner.plan(LocalRelation(Seq.empty[Attribute])).next()
+ withFallbackReason(op, "prior rule veto")
+ CometExplainInfo.withFallbackReasonScope {
+ withFallbackReason(expr, "current expression reason")
+ withFallbackReason(expr, "another expression reason")
+ val current = Set("current expression reason", "another expression reason")
+ assert(expr.getTagValue(tag).contains(current))
+ assert(CometExplainInfo.currentFallbackReasons(expr) == current)
+ withFallbackReason(op, "current operator reason")
+ assert(op.getTagValue(tag).contains(Set("prior rule veto", "current operator reason")))
+ }
+ withFallbackReason(expr, "after scope")
+ assert(
+ expr
+ .getTagValue(tag)
+ .contains(Set("current expression reason", "another expression reason", "after scope")))
+ }
+
+ test("issue #5499: child threads do not inherit expression fallback reasons") {
+ val source = Literal(true)
+ val reason = "parent conversion only"
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ CometExplainInfo.withFallbackReasonScope {
+ withFallbackReason(source, reason)
+ // Create the child while the parent scope is active: an inheritable ThreadLocal would
+ // expose the parent's journal and retain it beyond the parent's conversion.
+ val task = new FutureTask[Set[String]](() => {
+ CometExplainInfo.withExpressionFallbackReasonScope {
+ val owner = Alias(source, "child")()
+ QueryPlanSerde.liftFallbackReasons(source, owner)
+ owner.getTagValue(tag).getOrElse(Set.empty[String])
+ }
+ })
+ val thread = new Thread(task, "comet-fallback-provenance")
+ thread.setDaemon(true)
+ thread.start()
+ try assert(task.get(10, TimeUnit.SECONDS).isEmpty)
+ finally thread.join(10000)
+ val owner = Alias(source, "parent")()
+ QueryPlanSerde.liftFallbackReasons(source, owner)
+ assert(owner.getTagValue(tag).contains(Set(reason)))
+ }
+ }
+
+ test("issue #5499: aggregate preflight retains current reasons without importing stale tags") {
+ withObjectHashAggPlan { plan =>
+ val agg = partialObjectHashAgg(plan)
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ val expression = agg.aggregateExpressions.head
+ val key =
+ CometConf.getExprEnabledConfigKey(expression.aggregateFunction.getClass.getSimpleName)
+ val stale = "issue-5499-stale-aggregate-preflight"
+ expression.setTagValue(tag, Set(stale))
+ expression.aggregateFunction.setTagValue(tag, Set(stale))
+ val expected = s"Expression support is disabled. Set $key=true to enable it."
+ val rule = CometExecRule(spark)
+ withSQLConf(key -> "false") {
+ CometExplainInfo.withFallbackReasonScope {
+ val outer = Literal(true)
+ withFallbackReason(outer, "outer conversion")
+ assert(!rule.canAggregateBeConverted(agg, Partial))
+ assert(agg.getTagValue(tag).contains(Set(expected)))
+ assert(new ExtendedExplainInfo().generateExtendedInfo(agg).contains(expected))
+ assert(CometExplainInfo.currentFallbackReasons(outer) == Set("outer conversion"))
+ assert(CometExplainInfo.currentFallbackReasons(expression).isEmpty)
+ }
+ }
+ }
+ withObjectHashAggPlan { plan =>
+ val agg = partialObjectHashAgg(plan)
+ agg.aggregateExpressions.head.setTagValue(
+ CometExplainInfo.FALLBACK_REASONS,
+ Set("issue-5499-stale-successful-preflight"))
+ assert(CometExecRule(spark).canAggregateBeConverted(agg, Partial))
+ assert(agg.getTagValue(CometExplainInfo.FALLBACK_REASONS).isEmpty)
+ }
+ }
+
+ test("issue #5499: outermost conversion exceptions do not leak fallback reasons") {
+ val source = Literal(true)
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ intercept[IllegalArgumentException] {
+ CometExplainInfo.withFallbackReasonScope {
+ withFallbackReason(source, "failed conversion")
+ throw new IllegalArgumentException("conversion failed")
+ }
+ }
+ // Reuse the calling thread and source node after the exception, with its old tag intact.
+ CometExplainInfo.withExpressionFallbackReasonScope {
+ val owner = Alias(source, "next_conversion")()
+ QueryPlanSerde.liftFallbackReasons(source, owner)
+ assert(owner.getTagValue(tag).isEmpty)
+ withFallbackReason(source, "current conversion")
+ QueryPlanSerde.liftFallbackReasons(source, owner)
+ assert(owner.getTagValue(tag).contains(Set("current conversion")))
+ }
+ }
+
+ test("issue #5499: stale expression reasons cannot satisfy strict fallback checking") {
+ val literal = Literal.TrueLiteral
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ val previous = literal.getTagValue(tag)
+ val stale = "issue-5499-unrelated-query"
+ try {
+ literal.setTagValue(tag, Set(stale))
+ withTempView("fallback_provenance") {
+ createTestDataFrame.createOrReplaceTempView("fallback_provenance")
+ val sparkPlan = createSparkPlan(spark, "SELECT id * 2 FROM fallback_provenance")
+ 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)
+ def operator = ProjectExec(Seq(Alias(literal, "value")()), nativeChild)
+ CometExplainInfo.withFallbackReasonScope {
+ val op = operator
+ rule.rollUpFallbackReasons(op)
+ assert(op.getTagValue(tag).isEmpty)
+ intercept[IllegalStateException](rule.reportUnexplainedFallback(op))
+ withSQLConf(CometConf.COMET_STRICT_FALLBACK_REASONS.key -> "false") {
+ rule.reportUnexplainedFallback(op)
+ assert(op.getTagValue(tag).contains(Set(s"${op.nodeName} is not supported")))
+ }
+ // A copied tag is not a current reason, even if its source was tagged in this scope.
+ val source = Literal(true)
+ withFallbackReason(source, stale)
+ val copy = Literal(true)
+ assert(source == copy && !(source eq copy))
+ copy.copyTagsFrom(source)
+ val copied = Alias(copy, "copied")()
+ val copiedProject = ProjectExec(Seq(copied), nativeChild)
+ rule.rollUpFallbackReasons(copiedProject)
+ assert(copiedProject.getTagValue(tag).isEmpty)
+ intercept[IllegalStateException](rule.reportUnexplainedFallback(copiedProject))
+ }
+ // A current reason equal to the stale string must still count.
+ CometExplainInfo.withFallbackReasonScope {
+ val op = operator
+ withFallbackReason(literal, stale)
+ intercept[IllegalArgumentException] {
+ CometExplainInfo.withFallbackReasonScope {
+ withFallbackReason(literal, "nested conversion")
+ throw new IllegalArgumentException("conversion failed")
+ }
+ }
+ rule.rollUpFallbackReasons(op)
+ assert(op.getTagValue(tag).contains(Set(stale)))
+ rule.reportUnexplainedFallback(op)
+ }
+ CometExplainInfo.withFallbackReasonScope {
+ val op = operator
+ rule.rollUpFallbackReasons(op)
+ intercept[IllegalStateException](rule.reportUnexplainedFallback(op))
+ }
+ // Legitimate literal failures must survive both serialization and operator roll-up.
+ val unsupported =
+ Literal.create(new Object, org.apache.spark.sql.types.ObjectType(classOf[Object]))
+ val disabledKey = CometConf.getExprEnabledConfigKey("Literal")
+ Seq(unsupported -> Seq.empty[(String, String)], literal -> Seq(disabledKey -> "false"))
+ .foreach { case (value, configs) =>
+ withSQLConf(configs: _*) {
+ CometExplainInfo.withFallbackReasonScope {
+ val op = ProjectExec(Seq(Alias(value, "value")()), nativeChild)
+ assert(
+ QueryPlanSerde.exprToProto(op.projectList.head, nativeChild.output).isEmpty)
+ rule.rollUpFallbackReasons(op)
+ val expected = if (value eq literal) {
+ s"Expression support is disabled. Set $disabledKey=true to enable it."
+ } else {
+ s"Unsupported data type ${unsupported.dataType}"
+ }
+ assert(op.getTagValue(tag).contains(Set(expected)))
+ assert(new ExtendedExplainInfo().generateExtendedInfo(op).contains(expected))
+ rule.reportUnexplainedFallback(op)
+ }
+ }
+ }
+ }
+ }
+ } finally {
+ previous match {
+ case Some(reasons) => literal.setTagValue(tag, reasons)
+ case None => literal.unsetTagValue(tag)
+ }
+ }
+ }
+
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
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala
index 8181628242..68f88e1825 100644
--- a/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala
@@ -20,12 +20,12 @@
package org.apache.spark.sql.comet
import org.apache.spark.sql.CometTestBase
-import org.apache.spark.sql.catalyst.expressions.{Add, ArrayContains, AttributeReference, BitwiseNot, Cast, CreateArray, Divide, EvalMode, Multiply, NamedExpression}
+import org.apache.spark.sql.catalyst.expressions.{Add, ArrayContains, AttributeReference, BitwiseNot, Cast, CreateArray, Divide, EvalMode, If, Literal, Multiply, NamedExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Partial, Sum}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DecimalType, IntegerType}
-import org.apache.comet.CometConf
+import org.apache.comet.{CometConf, CometExplainInfo}
import org.apache.comet.serde.{CometDivide, ExprOuterClass, QueryPlanSerde, Unsupported}
class CometDecimalPromotionSuite extends CometTestBase {
@@ -37,6 +37,36 @@ class CometDecimalPromotionSuite extends CometTestBase {
private case class TestMode(name: String, ansiEnabled: Boolean, failOnError: Boolean)
+ test("issue #5499: decimal fallback lift ignores stale shared literal reasons") {
+ val literal = Literal.TrueLiteral
+ val tag = CometExplainInfo.FALLBACK_REASONS
+ val previous = literal.getTagValue(tag)
+ val stale = "issue-5499-unrelated-query"
+ try {
+ literal.setTagValue(tag, Set(stale))
+ val amount = AttributeReference("amount", DecimalType(10, 2))()
+ val expression = If(literal, Add(amount, amount), amount)
+ assert(!(DecimalPrecision.promote(expression) eq expression))
+ val key = CometConf.getExprEnabledConfigKey("Add")
+ withSQLConf(key -> "false") {
+ assert(QueryPlanSerde.exprToProto(expression, Seq(amount)).isEmpty)
+ assert(
+ expression
+ .getTagValue(tag)
+ .contains(Set(s"Expression support is disabled. Set $key=true to enable it.")))
+ }
+ // Reusing the shared literal in a later successful conversion must not lift its old tag.
+ val successful = If(literal, Add(amount, amount), amount)
+ assert(QueryPlanSerde.exprToProto(successful, Seq(amount)).isDefined)
+ assert(successful.getTagValue(tag).isEmpty)
+ } finally {
+ previous match {
+ case Some(reasons) => literal.setTagValue(tag, reasons)
+ case None => literal.unsetTagValue(tag)
+ }
+ }
+ }
+
test("issue #5190: decimal promotion is idempotent during recursive serialization") {
val left = "CAST(id AS DECIMAL(10, 0))"
val right = "CAST(id + 1 AS DECIMAL(10, 0))"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org