andygrove opened a new pull request, #5652:
URL: https://github.com/apache/datafusion-comet/pull/5652

   ## Which issue does this PR close?
   
   Closes #5618.
   
   ## Rationale for this change
   
   When a native Iceberg write task fails partway through, the data files it 
had already finalized stay in the table's data location. They are invisible to 
readers and `remove_orphan_files` eventually reclaims them, but iceberg-java 
deletes them synchronously (`DataWriter.abort()` calls 
`SparkCleanupUtil.deleteTaskFiles`), so on spot-heavy or preemption-prone 
clusters the native path silently accumulates orphans that the JVM path does 
not. This is the first phase-2 item of the native Iceberg writes epic (#5649).
   
   ## What changes are included in this PR?
   
   Two halves, matching the two places a task can fail.
   
   **Inside the native writer.** iceberg-rust's writers keep the `DataFile`s 
they have finalized private until `close` and have no abort hook, so the task 
cannot ask a failed writer what it wrote. Instead, the 
`RollingFileWriterBuilder` is given a `TrackingLocationGenerator`, a wrapper 
around `DefaultLocationGenerator` that records every location it hands to a 
file writer. If the input stream, a write, or the final close fails, 
`run_write_task` deletes every recorded location through the task's `FileIO` 
before propagating the original error.
   
   That explicit path is not enough on its own, and the new end-to-end test 
proved it: when the JVM-side input iterator throws (a UDF failure upstream of 
the write, the most common shape), `executePlan` returns the error straight 
from its JNI batch pull and the JVM releases the plan, so the write task's 
future is dropped without ever seeing an error. An `AbortOnDrop` guard covers 
that case. It is disarmed once the task completes; if it is dropped while still 
armed it deletes the tracked files, synchronously on a throwaway current-thread 
runtime when dropped from a plain JVM thread (`releasePlan`), or spawned onto 
the current runtime when dropped from inside one. Deletion is best-effort in 
both paths: failures are logged, never returned, so the task failure Spark 
reports is still the real one. The location that was open at the time of the 
failure is included; deleting a path that was never materialized is a no-op.
   
   **After the native writer has returned.** Once the manifest is decoded, the 
JVM knows the files. `CometIcebergWriteExec` registers a `TaskFailureListener` 
that deletes them through the table `FileIO` already in the task closure if the 
metrics rebuild, `TaskCommit` construction, or serialization fails. 
`IcebergReflection.deleteFilesQuietly` prefers 
`SupportsBulkOperations.deleteFiles` and falls back to 
`FileIO.deleteFile(String)` per path, and likewise never throws.
   
   The "Failure handling" section of `iceberg-writes.md` is updated to describe 
the cleanup and drop the reference to this issue.
   
   ## How are these changes tested?
   
   - Rust unit tests for the tracking generator (layout unchanged, every 
location recorded), for the cleanup against an in-memory `FileIO` including a 
recorded location that was never written, and for the drop guard (disarmed 
guard deletes nothing; armed guard dropped outside a runtime deletes 
synchronously).
   - `CometIcebergWriteActionSuite`: a failure-injection test writes a 
single-task source with a two-row Comet batch size into a table with a one-byte 
`write.target-file-size-bytes`, so the rolling writer finalizes a file per 
batch, and a UDF throws on the seventh row. A control run with the same source 
and settings first proves the writer rolls into several files; the failing run 
then asserts the write planned natively (`CometIcebergWriteExec` in the failed 
plan), no snapshot was created, the pre-existing data file is untouched, and no 
other parquet file remains under the table's data location.
   - A direct test of `deleteFilesQuietly` through a real table's 
`HadoopFileIO`: written files are removed, a nonexistent path is tolerated, and 
a second call is a no-op.
   - The existing Iceberg write suites were run 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]

Reply via email to