[osipovartem]

Summary

• Track every generated unpartitioned data-file path so failed writes can remove uncommitted objects, including failures without declared column limits.
• Clean up after sink and post-sink metadata errors; preserve the original write error and log cleanup failures.
• Return partitioned demux send errors instead of panicking when a writer closes.
• Add a late-input-error regression after a data file has materialized.

Dependency and merge gate

Keep this PR draft as a standalone fork change. Its Cargo.toml still resolves crates.io DataFusion 55.0.0, whose Parquet sink can return before its demux and writer tasks terminate on error. This cleanup could then delete a path before a late writer recreates it. Embucket/datafusion#167 fixes the sink, and Rustice pins both exact SHAs, so the paired Rustice integration is safe, but that pin does not protect other iceberg-rust consumers. Before merging this PR independently, ensure the fixed sink is guaranteed by a dependency all consumers actually resolve (or implement that guarantee in this fork); a workspace-only [patch.crates-io] does not propagate to downstream roots. The current late-input regression does not cover the writer-error race.

Verification

• Focused cargo +1.95.0 test -p datafusion_iceberg --lib late_input_error_cleans_previously_written_data_file_without_limits: 1 passed locally.
• cargo +1.95.0 fmt --all -- --check and git diff --check passed.
• Independent read-only review approved the paired Rustice integration; a follow-up dependency review blocked standalone merge until all consumers resolve a quiescent DataFusion sink.
• Rustice integration pins this revision plus the fixed DataFusion revision and passed the strict 541-file, 4195-case local SQLLogicTest corpus.

Performance

• Successful unpartitioned writes now record each generated file path once; no per-row work was added. The DataFusion sink dependency was benchmarked against its base revision in Drain Parquet writer tasks before returning failures datafusion#167 with a small measured local-write overhead (+1.08% for one file, +1.58% for 32 files in 15 paired large-input runs). A separate end-to-end Iceberg throughput benchmark was not run.
• The per-file bookkeeping and small sink overhead are accepted to ensure failed writes can remove all uncommitted files after all writer tasks stop. The independent follow-up review identified the unpinned crates.io DataFusion dependency as a blocking standalone-merge issue.