sunchao commented on code in PR #5696:
URL: https://github.com/apache/datafusion-comet/pull/5696#discussion_r3938095710
##########
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 =>
+ stripColumnarToRow(w.child).map(child =>
w.withNewChildren(Seq(child))).getOrElse(w)
Review Comment:
### Correctness
[P2] Preserve the Arrow bridge before visiting the writer
Could this protect the writer's input before the bottom-up cancellation
runs, while retaining the plain C2R handling needed by AQE? A direct leaf
converted by `CometSparkToColumnarExec.createExec` is initially a
`CometScanWrapper`, which extends `CometNativeExec` and passes
`requiresNativeChildren`. `CometExecRule` then removes the wrapper, leaving the
Arrow bridge directly below the writer. With this PR, Spark inserts
`ColumnarToRowExec(CometSparkToColumnarExec(source))`, and the existing arm at
lines 81–88 processes it before this writer arm. For a row source it removes
both transitions, so this arm receives a bare row child and cannot restore the
bridge. The writer then throws `CometIcebergWriteExec requires a columnar
(Comet native) child`.
This affects a direct `RangeExec`/RDD source with Spark-to-columnar
conversion enabled and that operator allowed, for example an unpartitioned
identity append without an intervening sort/exchange. The matched component
probe with a real `RangeExec` retains the bridge before the PR and loses it in
the proposed implementation. The new documentation commit leaves that
executable code unchanged. Both plans have only one transition, so this is
separate from the transition-heavy fallback concern. A Spark-columnar source
also loses its Arrow bridge, leaving non-`CometVector` input for the FFI
adapter even though `supportsColumnar` remains true. Please preserve that
conversion and cover both input representations in the regression tests.
--
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]
