jordepic commented on code in PR #5696:
URL: https://github.com/apache/datafusion-comet/pull/5696#discussion_r3937608443
##########
spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala:
##########
@@ -91,6 +91,18 @@ case class EliminateRedundantTransitions(session:
SparkSession)
// Write should be final operation in the plan
case ColumnarToRowExec(nativeWrite: CometNativeWriteExec) =>
nativeWrite
+ // `CometIcebergWriteExec` is row-based (it emits the serialised Iceberg
commit message) but
+ // consumes Arrow batches from its child over FFI, so Spark inserts a
columnar-to-row
+ // transition *underneath* it. Strip it so `doExecuteColumnar` sees the
columnar child
+ // directly; `CometIcebergNativeWrite.requiresNativeChildren` already
guarantees that child
+ // was Comet-native when the write was converted.
+ //
+ // The write deliberately does not tag itself as a
`ColumnarToRowTransition` to suppress the
+ // insertion: Spark leaves such a node untouched, so the whole subtree
below the write is
+ // never visited and the transitions the rest of that subtree needs are
never inserted
+ // (https://github.com/apache/datafusion-comet/issues/5689).
+ case w: CometIcebergWriteExec =>
Review Comment:
The comment above says the child was guaranteed Comet-native at conversion
time, but `RevertNativeForTransitionHeavyStages` runs between conversion and
this rule (it is first in `postColumnarTransitions`). That rule counts
`ColumnarToRowTransition` nodes in the stage, and the write itself used to
count as one, so the new `ColumnarToRowExec` underneath simply takes its place
and the count is unchanged. So no behaviour change from this PR there.
While looking at that interaction I noticed something pre-existing that this
PR does not introduce but sits right next to:
`CometIcebergWriteExec.originalPlan` is `child`, and `revertToSpark` replaces
every `CometExec` with `originalPlan.withNewChildren(children)`. If the write's
stage ever exceeds `maxTransitions` (default 2, reachable with
`spark.comet.sparkToColumnar.enabled` and, say, a row-based `Union` of two
Spark-columnar scans directly under the write), the write node disappears and
`IcebergCommitExec` would try to deserialise data rows as commit messages.
`CometNativeWriteExec` has the same `originalPlan = child`. Does that deserve a
tracking issue? It feels like a separate fix, but it is the same
transition-accounting area this PR touches, so I wanted to raise it here rather
than lose it.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1688,6 +1741,31 @@ class CometIcebergWriteActionSuite
assume(icebergAvailable, "Iceberg not available in classpath")
}
+ /**
+ * Every row-consuming operator must receive row-based input. Spark
guarantees that by inserting
+ * `ColumnarToRow` transitions in `ApplyColumnarRulesAndInsertTransitions`;
an operator that
+ * Comet rewrote in a way that skips the insertion pass shows up here as a
row-based node with a
+ * columnar-only child, and would fail at runtime with a `ColumnarBatch
cannot be cast to
+ * InternalRow` `ClassCastException` rather than at planning time.
+ *
+ * `CometIcebergWriteExec` is the one legitimate exception: it is row-based
on the outside but
+ * pulls Arrow batches from its Comet-native child over FFI (see the class
docstring), so the
+ * transition below it is deliberately stripped again by
`EliminateRedundantTransitions`.
+ */
+ private def assertColumnarContract(plan: SparkPlan): Unit = {
Review Comment:
This helper is a nice general guard for exactly the class of bug the issue
describes (a Comet rewrite hiding part of the subtree from Spark's insertion
pass), and `capturePlans` already records `qe.executedPlan` for every write in
the suite. Would it be worth calling `assertColumnarContract` from
`capturePlans` (or `captureWrite`) so all the existing tests check the contract
too, rather than only this one? That would also make it cheap to cover the
UPDATE and MERGE shapes from the issue with AQE off, since MERGE in particular
puts a different join and `MergeRows` between the columnar CoW scan and the
write.
--
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]
