andygrove commented on code in PR #5610:
URL: https://github.com/apache/datafusion-comet/pull/5610#discussion_r3970394929
##########
spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala:
##########
@@ -342,6 +342,84 @@ abstract class CometTestBase
(sparkPlan, cometPlan)
}
+ /**
+ * Check for the correct results, that Comet replaced all possible
operators, and that the named
+ * expressions ran through the mechanism the caller expects.
+ *
+ * Comet evaluates an expression one of three ways: natively (a DataFusion
expression), through
+ * the JVM codegen dispatcher (Spark's own `doGenCode` compiled into an
Arrow batch kernel), or
+ * not at all (the operator falls back to Spark). Only the third is visible
to
+ * [[checkSparkAnswerAndOperator]]; the first two produce Spark-matching
results by
+ * construction, so a serde that quietly widens from native to dispatch
(losing the native
+ * kernel) or narrows from dispatch to native (losing Spark-exact semantics)
passes every other
+ * assertion here. Use this to pin which one actually ran.
+ *
+ * Names are the expression's `prettyName` lowercased, as
[[ExtendedExplainInfo]] reports them
+ * (`bit_length`, `octet_length`, `rlike`), not necessarily the SQL alias
used to invoke it: a
+ * function registered with `setAlias` reports the invoked alias, everything
else reports its
+ * own `prettyName`.
+ *
+ * For fallback assertions use [[checkSparkAnswerAndFallbackReason]] instead.
+ */
+ protected def checkSparkAnswerAndImpl(
+ df: => DataFrame,
+ native: Seq[String] = Seq.empty,
+ dispatched: Seq[String] = Seq.empty): (SparkPlan, SparkPlan) = {
+ val (sparkPlan, cometPlan) = checkSparkAnswerAndOperator(df)
+ assertExpressionImpl(cometPlan, native, dispatched)
+ (sparkPlan, cometPlan)
+ }
+
+ /** Check for the correct results and the expected per-expression
implementation. */
+ protected def checkSparkAnswerAndImpl(
+ query: String,
+ native: Seq[String],
+ dispatched: Seq[String]): (SparkPlan, SparkPlan) = {
+ checkSparkAnswerAndImpl(sql(query), native, dispatched)
+ }
+
+ /**
+ * Assert how Comet evaluated the named expressions in an already-executed
Comet plan. Split out
+ * from [[checkSparkAnswerAndImpl]] so callers holding a plan can reuse it,
and so the assertion
+ * itself is testable.
+ *
+ * Each name must appear in its expected set and must be absent from the
other, so naming an
+ * expression is a claim about which mechanism ran it rather than a claim
that it ran somehow.
+ */
+ protected def assertExpressionImpl(
+ cometPlan: SparkPlan,
+ native: Seq[String],
+ dispatched: Seq[String]): Unit = {
+ val explainInfo = new ExtendedExplainInfo()
+ val actualNative = explainInfo.getNativeExpressions(cometPlan)
+ val actualDispatched = explainInfo.getCodegenDispatchExpressions(cometPlan)
+ def detail: String =
+ s"native=[${actualNative.mkString(", ")}] " +
+ s"codegen-dispatched=[${actualDispatched.mkString(", ")}]"
+ native.foreach { name =>
+ if (actualDispatched.contains(name)) {
Review Comment:
Fixed in b3fcf823e. You were right, and it was source-derivable — I
confirmed it by writing the case you described and watching it pass when it
should not have.
The cause is at the tagging site rather than in the exclusion check.
`emitJvmCodegenDispatch` binds and closure-serializes the *whole* subtree into
one kernel — its only inputs are the `AttributeReference`s the tree reads — but
it named only the root. So for `hypot(abs(b), c)` the dispatched set held
`hypot` alone, and the inner `abs` was in neither set.
Rather than teach the exclusion check about descendants, I made the
classification true at the source: the dispatch site now also names every
expression in the subtree. Attribute references and literals are left out,
since they are the kernel's inputs rather than work it performed (and would
otherwise show up in the coverage stats as "expressions").
That makes the helper reject both halves of your example, which is the point
— `abs` is genuinely on both sides of the fence in that query, so neither claim
about it alone is true:
```scala
val query = "SELECT abs(a), hypot(abs(b), c) FROM t"
intercept[TestFailedException] { checkSparkAnswerAndImpl(sql(query), native
= Seq("abs")) }
intercept[TestFailedException] { checkSparkAnswerAndImpl(sql(query),
dispatched = Seq("abs")) }
checkSparkAnswerAndImpl(sql(query), dispatched = Seq("hypot")) // still
unambiguous
```
That is *an expression nested inside a dispatched subtree is classified as
dispatched* in `CometCodegenSuite`. Against the previous head the first
`intercept` fails with "Expected exception ... but no exception was thrown", so
it does reproduce your case rather than just documenting it.
**Two consequences worth flagging, since this tag is not only used by the
test helper.**
The extended-explain line and the coverage stats read the same tag, so both
got more accurate: a query like `sequence(a, comet_seq_stopper())` now reports
`JVM codegen dispatcher: comet_seq_stopper, sequence` and counts 2 rather than
1. Previously the nested UDF was counted in neither bucket.
That broke two existing assertions in `CometCodegenSuite` that did
`explain.contains("JVM codegen dispatcher: sequence")` — a substring match
pinned to the name being first in a sorted list. I replaced that pattern
everywhere it appeared with a `dispatchedNames(explain)` helper that parses the
segment and checks membership, so it does not break again the next time a query
dispatches a second expression. Full `CometCodegenSuite` (95) and
`CometSqlFileTestSuite` (472) pass.
##########
spark/src/test/scala/org/apache/comet/CometSqlFileTestSuite.scala:
##########
@@ -155,6 +155,10 @@ class CometSqlFileTestSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
checkSparkAnswerAndOperatorWithTolerance(sql, tol)
case ExpectFallback(reason) =>
checkSparkAnswerAndFallbackReason(sql, reason)
+ case ExpectDispatch(names) =>
Review Comment:
Done in b3fcf823e — agreed, and the argument that they are *stronger*
controls is the right one: both branches call `checkSparkAnswerAndImpl`, which
runs the same answer and operator checks a plain `query` does before asserting
anything about the mechanism. Refusing them meant the preflight pushed fixtures
toward a weaker sentinel.
```scala
val hasSentinel = file.records.exists {
case SqlQuery(_, CheckCoverageAndAnswer | _: ExpectDispatch | _:
ExpectNative, _) => true
case _ => false
}
```
I also updated the failure message, which still told authors to add a plain
`query` — the advice would have been wrong as soon as this changed:
> Add at least one `query`, `expect_dispatch` or `expect_native` over valid
input so the operator check fails if the expression did not execute natively.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
