andygrove opened a new pull request, #5663: URL: https://github.com/apache/datafusion-comet/pull/5663
## 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. -- 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]
