mbutrovich commented on code in PR #6082:
URL: https://github.com/apache/datafusion-comet/pull/6082#discussion_r4065382220
##########
spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala:
##########
@@ -1086,4 +1086,30 @@ class CometExecRuleSuite extends CometTestBase {
}
}
+ test("operator conversion alone converts nothing without scan conversion") {
Review Comment:
This test passes only because `spark.comet.convert.parquet.enabled` defaults
to `false`
([`CometConf.scala#L180-L186`](https://github.com/apache/datafusion-comet/blob/09b44ad6fa17f58f4bbccf958c5cf02d790f7334/spark/src/main/scala/org/apache/comet/CometConf.scala#L180-L186)).
With it on, the same plan comes back with a `CometSparkToColumnarExec` over
the scan and native operators above it, so the name and the comment describe an
invariant that doesn't hold. It also tests `CometExecRule` in isolation and
never runs `CometRule`, which is the class this PR adds.
Could it test the property `CometRule` guarantees instead? Set
`COMET_CONVERT_FROM_PARQUET_ENABLED` to `false` explicitly so the test doesn't
lean on a default, assert that `CometExecRule` alone leaves the
`FileSourceScanExec` in place, and assert that `CometRule` on the same plan
produces a `CometNativeScanExec`. That fails if someone reorders or drops the
scan phase. The current test doesn't.
```scala
test("scan conversion must run before operator conversion") {
...
withSQLConf(
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "false") {
val plan = stripAQEPlan(sparkPlan)
assert(countOperators(CometExecRule(spark).apply(plan),
classOf[FileSourceScanExec]) == 1)
assert(countOperators(CometRule(spark).apply(plan),
classOf[CometNativeScanExec]) == 1)
}
}
```
##########
spark/src/main/scala/org/apache/comet/rules/CometRule.scala:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.rules
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.SparkPlan
+
+/**
+ * Comet's plan conversion pass: scan conversion followed by operator
conversion.
+ *
+ * The two were previously registered as separate rules, adjacently, in both
the columnar and the
+ * query-stage-prep paths. Nothing ever ran between them, and neither is
useful on its own -
+ * [[CometExecRule]] seeds its native chain only from the nodes
[[CometScanRule]] produces
+ * (`CometScanExec`, `CometBatchScanExec`, `CometContribScanMarker`), so
operator conversion
+ * against unconverted Spark scans converts nothing. Composing them here makes
that ordering an
+ * invariant of the code rather than of the registration order, and gives
callers that need the
+ * whole conversion - rather than half of it - a single entry point.
+ *
+ * The two rules keep their own classes, files and tests; this only fixes how
they are sequenced.
+ * `ruleName` in `spark.comet.explain.transformations` output is still each
inner rule's own,
+ * since this delegates to their `apply`.
Review Comment:
"Operator conversion against unconverted Spark scans converts nothing" isn't
accurate. `CometExecRule` wraps an unconverted `FileSourceScanExec` in
`CometSparkToColumnarExec` when `spark.comet.convert.parquet.enabled=true`
([`CometExecRule.scala#L411-L412`](https://github.com/apache/datafusion-comet/blob/09b44ad6fa17f58f4bbccf958c5cf02d790f7334/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala#L411-L412)),
and then converts the operators above it. It also converts local table scans,
empty relations and shuffles with no scan conversion at all. The suggestion
below states the narrower property.
The docstring's last sentence is true for
`spark.comet.explain.transformations`, but Spark's own plan change log now sees
one rule. `AdaptiveSparkPlanExec.applyPhysicalRules` calls
`logger.logRule(rule.ruleName, ...)` once per registered rule ([Spark
`AdaptiveSparkPlanExec.scala#L856-L860`](https://github.com/apache/spark/blob/303c18c74664f161b9b969ac343784c088b47593/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala#L856-L860)).
Anyone filtering `spark.sql.planChangeLog.rules` on
`org.apache.comet.rules.CometScanRule` or `CometExecRule` will stop seeing
output and must switch to `org.apache.comet.rules.CometRule`. The PR
description should mention this, and the docstring should say how the rule
appears in that log.
Most of this docstring is archaeology. "Previously registered as separate
rules, adjacently", "nothing ever ran between them" and "keep their own
classes, files and tests" describe what this PR changed, not the code. None of
it is needed to understand the current code, so it should go. It belongs in the
PR description and commit message.
```suggestion
* Native scans come only from the nodes [[CometScanRule]] produces
(`CometScanExec`,
* `CometBatchScanExec`, `CometContribScanMarker`), so [[CometExecRule]]
must run after it.
* Running [[CometExecRule]] alone leaves scans on Spark's readers.
Composing the two here makes
* that ordering part of the code instead of depending on the order the
rules are registered, and
* gives callers that need the whole conversion a single entry point.
*
* `spark.comet.explain.transformations` logs each inner rule under its own
`ruleName`. Spark's
* `spark.sql.planChangeLog.*` output logs this rule as a single step named
`CometRule`.
```
--
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]
