andygrove commented on code in PR #5696:
URL: https://github.com/apache/datafusion-comet/pull/5696#discussion_r3941229492
##########
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:
Confirmed on both counts.
On the transition accounting: agreed, the count is unchanged.
`RevertNativeForTransitionHeavyStages` runs first in `postColumnarTransitions`
and `countTransitions` counts `ColumnarToRowTransition` nodes, so the
`ColumnarToRowExec` Spark now inserts below the write takes the place the write
itself used to occupy. No behaviour change from this PR there.
On `originalPlan`: I read `revertToSpark` again and it is worse than a
disappearing node. The arm is
```scala
case cometExec: CometExec =>
if (cometExec.originalPlan.children.size == cometExec.children.size) {
cometExec.originalPlan.withNewChildren(cometExec.children)
} else { ... }
```
For `CometIcebergWriteExec`, `originalPlan` is `child` and `children` is
`Seq(child)`, so when the child is itself unary the size check passes and the
result is `child.withNewChildren(Seq(child))`: the write is gone and the child
is duplicated beneath itself. `CometNativeWriteExec` has the same `originalPlan
= child`. `IcebergCommitExec` would then deserialise data rows as commit
messages, as you say.
This is pre-existing and orthogonal to the transition fix, so I have left it
out of this PR. It needs its own issue and fix; the shape of the fix is
probably that a write exec should not report its child as `originalPlan`, or
that `revertToSpark` should decline to revert a node whose `originalPlan` is
one of its own children.
--
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]
