apache / apache/datafusion-comet

revertToSpark erases CometIcebergWriteExec / CometNativeWriteExec because originalPlan is the node's own child

Open
#5,719 1 comment 0 reactions 1 assignee Claimed by @sam-1112 View on GitHub
bug requires-triage
Dominant language
Scala
Stars
1.3k
Forks
373
Avg merge
2d 4h
Merged PRs (30d)
198

Description

### Describe the bug

`RevertNativeForTransitionHeavyStages.revertToSpark` erases the write node when it reverts a stage containing `CometIcebergWriteExec` or `CometNativeWriteExec`, because both report their own child as `originalPlan`.

The revert arm is:

```scala
val reverted = transformStageUp(stripped) { case cometExec: CometExec =>
if (cometExec.originalPlan.children.size == cometExec.children.size) {
cometExec.originalPlan.withNewChildren(cometExec.children)
} else {
logWarning(...)
cometExec.originalPlan
}
}
```

For every other `CometExec` this is a like-for-like swap, because `originalPlan` is the Spark operator the Comet node replaced. The two write execs are different:

- `CometIcebergWriteExec.originalPlan` is `child` (`CometIcebergWriteExec.scala:73`)
- `CometNativeWriteExec.originalPlan` is `child` (`CometNativeWriteExec.scala:80`)

So `originalPlan.withNewChildren(cometExec.children)` reduces to `child.withNewChildren(Seq(child))`. The write node disappears, and whichever branch is taken depends only on the child's arity:

- **Leaf or multi-child child** (arity mismatch): the warning branch returns the child as-is. The write is gone.
- **Unary child** (sizes both 1): the child is grafted onto itself, so the write is gone *and* the child operator is duplicated.

Downstream, `IcebergCommitExec` calls `child.executeCollect()` and deserialises each row's binary column as a `WriterCommitMessage`. After the revert it is handed data rows with the table's own schema instead of the single `iceberg_commit_message` binary column, so no data is written and the commit either fails to deserialise or commits garbage.

The duplication is a second, independent defect: it double-applies the operator. Harmless for an idempotent `Filter`, not harmless for a projection carrying `monotonically_increasing_id()`, a sample, or a limit.

### Steps to reproduce

Calling `revertToSpark` directly on a write plan shows both shapes. With a leaf child:

```
=== INPUT ===
CometIcebergWrite [iceberg_commit_message#0], , ICEBERG_WRITER_UNPARTITIONED
+- CometSparkRowToColumnar
+-

=== REVERTED ===

write present: false
```

With a unary Comet child:

```
=== INPUT ===
CometIcebergWrite [iceberg_commit_message#6], , ICEBERG_WRITER_UNPARTITIONED
+- CometFilter [_1#4], (isnotnull(_1#4) AND (_1#4 > 2))
+- CometNativeScan parquet [_1#4] ...

=== REVERTED ===
Filter (isnotnull(_1#4) AND (_1#4 > 2))
+- Filter (isnotnull(_1#4) AND (_1#4 > 2))
+- ColumnarToRow
+- FileScan parquet [_1#4] ...

write present: false
```

Note the `iceberg_commit_message` output column is gone in both, and the `Filter` is duplicated in the second.

To reach this through a real query rather than a direct call, the write's stage has to exceed the transition threshold:

- `spark.comet.exec.transitionRevert.enabled=true` (default `false`, so this is opt-in)
- more than `spark.comet.exec.transitionRevert.maxTransitions` (default 2) columnar-to-row transitions between the write and the first exchange below it, for instance with `spark.comet.sparkToColumnar.enabled` and a row-based `Union` of Spark-columnar scans directly under the write

### Expected behavior

Reverting a stage should either leave the write node in place (only the operators below it are candidates for reverting) or replace it with the JVM write operator it stands in for, which for `CometIcebergWriteExec` is `IcebergWriteExec`. It should never erase the node that produces the commit-message rows, and it should never duplicate a child.

Two candidate fixes:

1. Stop `revertToSpark` from reverting a node whose `originalPlan` is one of its own children. That is a general guard: `originalPlan = child` means "I have no distinct Spark original", and grafting a child onto itself is never correct.
2. Give the write execs a real `originalPlan`. `CometIcebergWriteExec` was built from an `IcebergWriteExec`, so it could carry that node instead of its child.

### Additional context

Found while reviewing #5696 (credit to @jordepic for spotting it there). It is pre-existing and unrelated to that PR's transition fix: `countTransitions` counts `ColumnarToRowTransition` nodes, and the `ColumnarToRowExec` #5696 causes Spark to insert below the write simply takes the place the write node itself used to occupy in that count, so the revert decision is unchanged either way.

Part of the native Iceberg writes epic #5649.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.