fix: Delete completed tasks' data files when an Iceberg write job fails

[andygrove]

Which issue does this PR close?

Closes #5277.

Stacked on #5652 (task-level cleanup); the first commit here is that PR's commit and this PR's own change is the second one. It reuses the IcebergReflection.deleteFilesQuietly helper introduced there.

Rationale for this change

When one task of a multi-task Iceberg write fails, the tasks that had already completed leave their data files in the table's data location. IcebergCommitExec collected task commit messages with executeCollect, which only returns once every task has succeeded, so on a job failure the committer had no messages and aborted with an empty list. Together with #5652 (which cleans up the failing task's files) this closes the last way a failed write job leaves data files behind.

What changes are included in this PR?

IcebergCommitExec.collectAndCommit now runs the write job with sparkContext.runJob and a per-partition result handler, recording each task's commit message as that task finishes, the way Spark's own WriteToDataSourceV2Exec.writeWithV2 does. On a job failure it aborts with the messages of the completed tasks.

Passing those messages to Iceberg's abort turns out not to be enough. SparkWrite.abort deletes the files listed in the messages only when the preceding commit failed with a CleanableFailure; before any commit has been attempted its cleanupOnAbort flag is still false and it logs "Skipping cleanup of written files". That is a defensible choice for an unknown commit outcome, but after a job failure no commit was attempted and nothing can reference the files, so the committer now deletes the completed tasks' data files itself through the table FileIO, best-effort and after the abort. A new IcebergReflection.taskCommitFileLocations reads the data files out of SparkWrite$TaskCommit (package-private files()).

This applies to both the native writer and the JVM writer under the split-operator plan; the stock Spark path still leaves those files for remove_orphan_files. The failure-handling section of iceberg-writes.md is updated accordingly.

How are these changes tested?

Two new tests in CometIcebergWriteActionSuite, one with the native writer and one with the JVM writer: a three-task write (one task per source parquet file) where the failing task's UDF blocks on a SparkListener-driven latch until the other two tasks have finished, then throws. That guarantees the driver holds two completed commit messages when the job fails, so the assertions (no snapshot created, the pre-existing data file untouched, no other parquet file left under the table's data location, the gate saw two completed tasks) exercise the committer's cleanup rather than the task-level one.

The Iceberg write action, write detection, and rewrite action suites pass locally on the default Spark profile.

[sunchao]

Reviewed 01c9101eab180c2301455c5b50d1d68d345cd72a against base ef62b46306e925bc51e7d7f29922c1870eb729e7, including the stacked task-cleanup change. The indexed completion callback addresses cleanup of successful tasks when a sibling fails. Keeping unconditional deletion before the commit attempt also preserves unknown-commit handling. I found no additional P1/P2 in the driver delta, but one existing P2 remains in the submitted stack.

[P2] Retain cleanup ownership through JVM manifest decoding

Could we retain an independent cleanup owner until decoding and failure-listener registration succeed? This is the existing #5652 handoff discussion, not a new driver regression. In CometIcebergWriteExec.scala:184-194, native output is drained and the manifest decoded before the listener is registered. Native EOF has released the plan after its cleanup guard was disarmed. If decoding fails, this task never sends a commit message to the new driver callback, leaving its files outside both cleanup paths.

The earlier dependency-level control used a valid 864,547-byte manifest from 4,096 real Parquet files. Actual Iceberg 1.11 decoding failed under constrained remaining heap, leaving those files. A retained-guard control deleted all 4,096 files afterward. Source and dependency identities were reverified here, but that experiment was not rerun. It was not a full Spark/JNI reproduction, and real-workload frequency was not measured. The decoder failure predates this feature. The issue is incomplete cleanup ownership around the demonstrated failure.

Validation

I independently reran all eight component checks for the extracted commit/cleanup methods. They passed with real Iceberg messages and in-memory FileIO, using an explicit scheduler/batch-state simulation. Both new abort tests also passed in observed Spark 3.5 and 4.1 synthetic-merge CI. Current head checks show 65 successes and nine skips. No local full Spark/JNI query, Maven suite, or performance benchmark was run.

[sunchao]

Re-reviewed 2f65f57d16347e7940d620d932881fc2b5e20712 against 7f1e00189b1ed86f1cb5acd872d97fce694482b1, following the prior review. One new P1 prevents JVM test compilation, and the previously reported handoff P2 remains.

Correctness

[P1] Finish adapting the cleanup tests to the shared plan helper

The merged base moved capturePlans to CometIcebergTestBase.capturePlans(spark)(action), but the new control at line 1325 still uses the old call shape. The new captureFailedPlans helper also references QueryExecutionListener and QueryExecution without importing them. These are current compiler failures, not unexecuted test expectations: Spark 3.5 scans, Spark 4.1 scans, and Iceberg 1.11 report the same four errors. Update the call and restore the listener/type imports so JVM test compilation can complete.

[P2] Keep cleanup ownership through manifest decoding

The existing handoff finding remains unchanged. run_write_task disarms its guard before returning the data files; drainAvroPayload reaches native EOF, and JVM decoding at lines 190–192 precedes failure-listener registration at lines 197–208. A decoder failure therefore still occurs before either the task listener or driver receives ownership of those files. The guard/write-task body and JVM handoff block are identical to the previously reviewed head. The earlier dependency-level decoder control was not rerun; this is a source revalidation of the existing finding, with no duplicate inline comment.

Current CI has 12 successful, 38 failed, and 11 skipped checks. The inspected jobs check out 5de626f761645e2c6804a77cf9e7e42663de9632, whose parents are exactly this base/head and whose tree equals the head. Rust CI passed 1,186 tests with four skipped, including all three native cleanup unit tests. The inspected JVM jobs stopped at compilation. No local full Spark/JNI execution, component-harness rerun, or benchmark was performed. Maintained Spark 3.5/4.0 sources were checked; maintained 3.4/4.1 branches were unavailable.

Performance

The driver and deletion helpers are unchanged from the prior head. Result collection remains proportional to task messages, and native location tracking adds one stored path per file. Cleanup uses bulk deletion when available and otherwise deletes paths sequentially. No new performance improvement is claimed or measured in this follow-up.

Design

The indexed result callback still gives the driver completed messages before a sibling fails. On the maintained Spark 3.5/4.0 paths, AQE's execute() returns the final writer RDD; it does not collect the write's messages before this callback is installed. Unconditional file deletion remains confined to failure before the commit attempt. Commit exceptions retain Iceberg's abort policy, and successful commit/cache-refresh handling introduces no new deletion path.

Abstraction & complexity

The existing reflection helpers keep Iceberg-specific file extraction and deletion out of the committer. No additional abstraction is needed for this update. The remaining ownership boundary and incomplete test-helper migration are the concrete changes needed before this stack is ready.

[sunchao]

Correctness

[P1] Adapt the new cleanup tests to the shared plan helper

The merged base now defines capturePlans(spark: SparkSession)(action: => Unit) in CometIcebergTestBase, so this old call shape supplies the INSERT DataFrame where the SparkSession is required. The new captureFailedPlans helper at lines 1962–1964 also uses QueryExecutionListener and QueryExecution after their imports were removed. Please pass spark here and restore those imports. Current Spark 3.5 scans and Spark 4.1 scans both fail test compilation with these same four errors, so the cleanup tests cannot run. I verified the CI merge has the exact assigned base/head parents and the same source tree as this head.

[andygrove]

Fixed in a34fda4. All four were exactly as you described — this is the merge hazard where the base moves a helper into the shared trait and the branch's own new code still uses the old shape.

• capturePlans(spark) { ... } at the control-plan site, which was passing the INSERT DataFrame into the SparkSession parameter.
• Restored QueryExecution (added to the existing org.apache.spark.sql.execution import) and org.apache.spark.sql.util.QueryExecutionListener, which captureFailedPlans still needs even though capturePlans no longer does.

Verified by compiling both profiles rather than just the default: -Pspark-3.5 and the default 4.1 both now produce CometIcebergWriteActionSuite.class. The cleanup tests themselves run green:

- native acceleration: a failed task deletes the data files it already finalized
- a failed write job deletes the data files of tasks that completed (native=true)
- a failed write job deletes the data files of tasks that completed (native=false)
Tests: succeeded 3, failed 0

Separately: this branch now conflicts with main in native/core/src/execution/operators/iceberg_write.rs. I have deliberately not resolved that here, because it overlaps the abort-guard change in #5652 that this is stacked behind and your open P2 there moves the disarm point. Resolving it before that lands would mean merging a version of the guard that is about to change. I'll take the merge once #5652 settles.

[sunchao]

Correctness

Follow-up at 9a88c220dfa314a31972a79e11c9fc21456e56f0 against 424c31aa79d13fddf743ffa29bae3c6f146e6c5e, since review 5137168020.

The prior [P1] is addressed. The test uses capturePlans(spark), and captureFailedPlans now lives in the shared base with the required imports. Current-tree CI successfully compiles the test sources for Spark 3.5 / Scala 2.12 and Spark 4.0.

[P2] The existing cleanup handoff finding is narrowed but remains. Packaging errors now retain native cleanup, and manifest decoding is protected after the JVM adopts the independent location list. However, the native guard disarms before JNI export/Arrow import and before cleanup.own(decodeLocations(batch.column(1).getBinary(0))). That argument must finish copying and decoding the entire list before own runs. If, for example, the valid locations buffer cannot be allocated on the JVM heap, the listener still owns Nil, the native guard is gone, and the driver has no commit message for these files. Closing the batch or plan only releases execution resources. Please keep an independent cleanup owner until the JVM has recorded the locations, and exercise a failure during that handoff. This is source-verified. I did not run a new Spark/JNI or constrained-heap reproduction. No duplicate inline is proposed.

The exact-tree native CI job passed 1,264 tests with five skipped, including the abort-cancellation and locations-framing tests. The Java jobs above compile tests but skip their execution. Their checkout ffa6a322 has exactly this base/head as parents and the same whole tree as the reviewed head. At September 9, 17:35 UTC, CI had 23 successful checks, 7 skipped, 36 running and 1 queued, with no reported failure. Full CI is still pending. No local product test or new handoff fault injection was run. Maintained Spark 3.4/4.1 sources remain unavailable, so this review does not source-qualify those versions.

Performance

The additional locations column trades a second per-file path representation for avoiding the substantially larger manifest decode during cleanup adoption. The native tracker acquires its mutex when a file location is generated, not once per row. Encoding, the JVM binary copy and decoded strings all scale with file count and path length, so the new success-path overhead is concentrated in writes producing many small files. I found no new P1/P2 performance issue, and no benchmark was run. These source-level costs are not a measured regression or speedup.

Design

The separate locations column removes the dependency on successful manifest decoding, and leaving the native guard armed through packaging correctly protects those errors. The remaining ownership transfer must cover delivery and adoption too. Producing a stream item is not acknowledgement that the JVM has retained the locations. The current listener unit test exercises deletion after an explicit own, but does not exercise a failure before that adoption.

The driver operator is byte-identical to the previously reviewed head. Its indexed completion callback retains each completed task message before the job waiter reports success, matching the maintained Spark 3.5/4.0 ordering. Explicit deletion remains confined to failed runJob. Commit exceptions continue through Iceberg's abort policy, without adding unconditional deletion after a possibly successful commit.

Abstraction & complexity

The native guard, JVM task listener and Iceberg reflection helpers separate failure ownership, lifecycle notification and version-specific file deletion. The two-column payload is a justified separation from Avro decoding. The remaining P2 calls for a clear acknowledgement boundary between these existing owners, rather than more manifest-dependent cleanup logic. Moving captureFailedPlans into the shared test base also removes the duplicated helper and aligns the new tests with the current base.

[sunchao]

Correctness

I need to correct the P2 classification in my preceding review. I rechecked the complete discussion on the stacked #5652 change and my earlier assessment. The residual allocation failure before the JVM owns the locations was explicitly acknowledged as a limitation of best-effort cleanup. The demonstrated Avro-decoder and native packaging gaps have been fixed.

The conditional allocation failure I described remains possible, but I did not establish a new regression or reproduce a new failure in that smaller interval. I should not have carried the original P2 forward as though its demonstrated failure were still present. I withdraw that P2 classification and the request for a new JNI acknowledgement protocol. The P1 compilation issue is also fixed. I found no remaining verified P1/P2 on 9a88c220dfa314a31972a79e11c9fc21456e56f0, so I am approving this head.

The previously verified matching-tree native job passed 1,264 tests with five skipped, and matching-tree Spark 3.5/4.0 jobs compiled the JVM tests. At September 9, 17:46 UTC, CI had 35 successful checks, 7 skipped, 24 running and 1 queued, with no reported failure. Full CI remains pending. This correction adds no new test execution, benchmark or compatibility claim.

Performance

The path-tracking and locations-payload costs are unchanged from the preceding assessment and remain proportional to file count and path bytes. No new performance evidence justifies expanding this update to another cross-JNI protocol.

Design

The separate locations payload addresses the demonstrated dependency on successful manifest decoding. That bounded improvement should be assessed against its best-effort cleanup contract, without treating every possible resource-exhaustion failure as the original unresolved bug. The driver cleanup and uncertain-commit handling are unchanged.

Abstraction & complexity

The existing native guard, task listener and reflection helpers remain appropriately scoped for this change. No additional abstraction is requested on the basis of the withdrawn finding. This correction supersedes that classification, not the explicit validation limits.