andygrove commented on code in PR #5652:
URL: https://github.com/apache/datafusion-comet/pull/5652#discussion_r3944280493
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -357,19 +478,29 @@ async fn run_write_task(
// Build the field-id-decorated target schema once per task; every batch
is cast against it.
let target_schema =
Arc::new(iceberg::arrow::schema_to_arrow_schema(&iceberg_schema).map_err(iceberg_err)?);
- while let Some(batch) = input.try_next().await? {
- let decorated = decorate_batch_with_field_ids(batch, &target_schema)?;
+ let outcome = async move {
+ while let Some(batch) = input.try_next().await? {
+ let decorated = decorate_batch_with_field_ids(batch,
&target_schema)?;
+ let _timer = write_time.timer();
+ writer
+ .write(
+ decorated,
+ fanout_splitter.as_ref(),
+ clustered_splitter.as_ref(),
+ )
+ .await?;
+ }
let _timer = write_time.timer();
- writer
- .write(
- decorated,
- fanout_splitter.as_ref(),
- clustered_splitter.as_ref(),
- )
- .await?;
+ writer.close().await
}
- let _timer = write_time.timer();
- writer.close().await
+ .await;
+ // Whether the input stream, a write, or the close failed, every file this
attempt created is
+ // orphaned from here on: nothing will commit it, and the retry uses
attempt-unique names.
+ if outcome.is_err() {
+ delete_task_files(&file_io, location_generator.locations()).await;
+ }
+ abort_guard.disarm();
Review Comment:
Good catch, you're right — that window was real. `run_write_task` now
returns the still-armed guard alongside the `DataFile`s, exactly as you
suggested, and the outer task disarms it only after `build_output_batch` has
produced the batch. If `encode_data_files_as_manifest` or `build_output_batch`
fails in between, the task awaits `abort_guard.abort()` (delete-then-disarm)
before propagating the error, so the deletes finish rather than racing the
runtime teardown on the `Drop` path.
`a_successful_write_returns_an_armed_guard_that_can_still_delete_its_files`
pins the new contract: a successful write hands back a guard that is still
armed and whose recorded locations match the files on disk, and aborting it
removes them.
--
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]
