[andygrove]

Which issue does this PR close?

Part of #2827

Rationale for this change

This PR is another incremental step towards supporting native Parquet writes.

What changes are included in this PR?

File Commit Protocol Implementation Status

✅ Implemented:

Protobuf & Native Code:

• Added work_dir, job_id, task_attempt_id fields to ParquetWriter message
• Rust ParquetWriterExec accepts and uses commit protocol parameters
• Files written to staging directory (work_dir) instead of final location
• Task-specific filenames with attempt IDs (e.g., part-00000-00001.parquet)
• DataFusion metrics for write statistics (files_written, bytes_written, rows_written)

JVM Commit Protocol Integration:

• SQLHadoopMapReduceCommitProtocol instantiated in CometDataWritingCommand
• Job-level: setupJob(), commitJob(), abortJob() called on driver
• Task-level: setupTask(), commitTask(), abortTask() called per partition
• Automatic task commit/abort via iterator wrapper on completion/failure
• Native operator modified per-task with staging directory and task IDs
• Statistics logged after successful commit

Commit message collection

• Collected from executors using an accumulator

Error Handling:

• Task abort on iterator failure
• Job abort if commit fails
• Proper exception propagation

❌ Still TODO:

Feature Completeness:

• Dynamic partitioning not supported
• Bucketing not supported
• Only local filesystem (file://) paths supported (no HDFS, S3, etc.)

Validation:

• No comprehensive test coverage yet
• Need to verify commit protocol works correctly with task retries
• Need to test failure scenarios (partial write, task failure, job abort)

How are these changes tested?

Existing UT passes, and I see file committer log messages.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 57.69231% with 44 lines in your changes missing coverage. Please review.
✅ Project coverage is 59.15%. Comparing base (f09f8af) to head (f3b89a6).
⚠️ Report is 730 commits behind head on main.

Files with missing lines	Patch %	Lines
.../apache/spark/sql/comet/CometNativeWriteExec.scala	53.19%	36 Missing and 8 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #2828      +/-   ##
============================================
+ Coverage     56.12%   59.15%   +3.02%     
- Complexity      976     1486     +510     
============================================
  Files           119      167      +48     
  Lines         11743    15282    +3539     
  Branches       2251     2531     +280     
============================================
+ Hits           6591     9040    +2449     
- Misses         4012     4949     +937     
- Partials       1140     1293     +153     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[andygrove]

Thanks for the feedback so far @wForget. I have pushed some updates. Could you take another look when you have time?

[wForget]

The commit protocol class is configurable in spark. We should add a fallback when the configuration is not SQLHadoopMapReduceCommitProtocol.

https://github.com/apache/spark/blob/4435a3a7f0883559de3bd13ba84f311200943622/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L2194-L2203

[peter-toth]

+1

[wForget]

Thanks @andygrove , overall looks good to me

[andygrove]

Thanks @wForget @peter-toth @mbutrovich!