andygrove commented on PR #7043:
URL:
https://github.com/apache/datafusion-comet/pull/7043#issuecomment-6216921839
Thanks for splitting this guard out. The fallback test is skipped on every
CI profile, because upstream Iceberg has no `SortOrderAnalyzer`, so nothing in
CI reaches the `scanExec.ordering` check. Could we also add a test that doesn't
need Iceberg to report an ordering?
`BatchScanExec` is a case class, so the test can plan an ordinary Iceberg
query, `copy` the scan with an `ordering`, and run `CometScanRule` on it. Here
is one that does that. It goes at the end of `CometIcebergNativeSuite`, plus
two import changes: `{Ascending, DynamicPruningExpression, SortOrder}` from
`catalyst.expressions`, and `org.apache.comet.rules.CometScanRule`.
<details>
<summary>Suggested test</summary>
```scala
test("CometScanRule falls back for an Iceberg scan that carries an
ordering") {
assume(icebergAvailable, "Iceberg not available in classpath")
withTempIcebergDir { warehouseDir =>
withSQLConf(
"spark.sql.catalog.sort_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
"spark.sql.catalog.sort_cat.type" -> "hadoop",
"spark.sql.catalog.sort_cat.warehouse" ->
warehouseDir.getAbsolutePath,
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
val table = "sort_cat.db.ordered_scan_table"
try {
spark.sql(s"CREATE TABLE $table (id INT, name STRING) USING
iceberg")
spark.sql(s"INSERT INTO $table VALUES (3, 'c'), (1, 'a')")
// Plan the query with Comet off, then give the Iceberg scan the
ordering that a
// reporting Iceberg runtime would attach. No Iceberg release
reports one yet, so this is
// the only way to reach the guard on every profile.
var sparkPlan: SparkPlan = null
withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
sparkPlan =
stripAQEPlan(spark.sql(s"SELECT * FROM
$table").queryExecution.executedPlan)
}
val scan = collect(sparkPlan) { case b: BatchScanExec => b }.head
// Apply the rule to the scan alone. BatchScanExec.equals ignores
`ordering`, so
// TreeNode.transform would treat the copy as unchanged and keep
the original.
def withOrdering(ordering: Option[Seq[SortOrder]]): SparkPlan =
scan.copy(ordering = ordering)
val ordered = Some(Seq(SortOrder(scan.output.head, Ascending)))
// CometScanRule marks an accepted Iceberg scan as a
CometBatchScanExec. CometExecRule
// turns that into the native scan later. The unmodified scan and
one with an empty
// ordering are both accepted, so the assertion below is not
passing for some unrelated
// reason.
for (ordering <- Seq(None, Some(Seq.empty[SortOrder]))) {
val transformed =
CometScanRule(spark).apply(withOrdering(ordering))
assert(
collect(transformed) { case s: CometBatchScanExec => s
}.length == 1,
s"Expected the scan to be accepted for ordering $ordering.
Plan:\n$transformed")
}
val input = withOrdering(ordered)
assert(
collect(input) { case b: BatchScanExec => b.ordering
}.head.exists(_.nonEmpty),
s"Expected the copied scan to carry an ordering. Plan:\n$input")
val transformed = CometScanRule(spark).apply(input)
assert(
collect(transformed) { case s: CometBatchScanExec => s }.isEmpty,
s"Expected the ordered scan to stay on Spark.
Plan:\n$transformed")
assert(collect(transformed) { case b: BatchScanExec => b }.length
== 1)
val reasons = new
ExtendedExplainInfo().getFallbackReasons(transformed)
assert(
reasons.exists(_.contains("Iceberg reports an output ordering")),
s"Expected the ordering fallback reason but found: $reasons")
} finally {
spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
}
}
}
}
```
</details>
A few things I checked while writing it:
- It passes on the 3.4, 4.0 and 4.1 profiles.
- With the guard disabled (`if (false && scanExec.ordering...`) it fails
with "Expected the ordered scan to stay on Spark".
- The rule is applied to the scan alone on purpose. `BatchScanExec.equals`
ignores `ordering`, so `plan.transform { case b: BatchScanExec =>
b.copy(ordering = ...) }` treats the copy as unchanged and silently keeps the
original scan. My first version did that and the check passed for the wrong
reason. The test asserts that the copied scan carries the ordering, to catch
this.
- `CometScanRule` returns a `CometBatchScanExec` for an accepted scan.
`CometExecRule` turns that into `CometIcebergNativeScanExec` later, so the
assertions look for `CometBatchScanExec`.
The end-to-end test you already have can stay as the check on runtimes that
do report an ordering.
--
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]
