fix: Delete completed tasks' data files when an Iceberg write job fails - #5663
fix: Delete completed tasks' data files when an Iceberg write job fails#5663andygrove wants to merge 2 commits into
Conversation
iceberg-java's writer abort deletes the files a failed task attempt wrote; the native path left them for remove_orphan_files. Close the gap in both places a task can fail. Inside the native writer, a TrackingLocationGenerator records every location handed to a file writer, since iceberg-rust's writers keep finalized files private until close and have no abort hook. The task deletes the recorded locations when a write fails, and an AbortOnDrop guard does the same when the task future is dropped without ever seeing an error, which is what happens when the JVM input iterator throws: executePlan returns that error from its JNI batch pull and the JVM releases the plan. After the native writer has returned, CometIcebergWriteExec registers a task failure listener that deletes the decoded manifest's files through the table FileIO, via a new best-effort IcebergReflection helper. Both deletions log failures rather than raising them, so the original task failure is the one Spark reports. Closes apache#5618
IcebergCommitExec collected task commit messages with executeCollect, which only returns once every task has succeeded, so a job failure left the committer with no messages and the data files of the tasks that had completed stayed in the table's data location. Collect each task's message as it finishes through a runJob result handler, as Spark's own V2 write does, and abort with the completed messages. Iceberg's SparkWrite.abort only deletes files after a cleanable commit failure and skips cleanup before any commit was attempted, so the committer then deletes the completed tasks' data files itself through the table FileIO; nothing can reference them at that point. The failing task's own files are handled by task-level cleanup. Closes apache#5277
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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.
2 participants
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.deleteFilesQuietlyhelper 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.
IcebergCommitExeccollected task commit messages withexecuteCollect, 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.collectAndCommitnow runs the write job withsparkContext.runJoband a per-partition result handler, recording each task's commit message as that task finishes, the way Spark's ownWriteToDataSourceV2Exec.writeWithV2does. 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.abortdeletes the files listed in the messages only when the preceding commit failed with aCleanableFailure; before any commit has been attempted itscleanupOnAbortflag 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 tableFileIO, best-effort and after the abort. A newIcebergReflection.taskCommitFileLocationsreads the data files out ofSparkWrite$TaskCommit(package-privatefiles()).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 oficeberg-writes.mdis 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 aSparkListener-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.