fix: wrap Iceberg split-write failures the way Spark does when abort fails

[andygrove]

Which issue does this PR close?

Closes #6143.

Rationale for this change

#6143 reported that IcebergCommitExec rethrows the raw cause of a failed write where Spark 3.x wraps it in a "Writing job aborted" SparkException. Checking Spark's source shows that premise is only true for Spark 3.3 and earlier, which Comet does not support.

At every Spark version Comet builds against (3.4, 3.5, 4.0, 4.1, 4.2), V2TableWriteExec.writeWithV2 catches any Throwable from the write job or from batchWrite.commit and aborts. Then:

• if batchWrite.abort succeeds, it rethrows the original cause unchanged
• if batchWrite.abort itself throws, it attaches the abort failure to the cause as suppressed and throws QueryExecutionErrors.writingJobFailedError(cause), a SparkException ("Writing job failed.") with the original failure as its cause

Spark tag	WriteToDataSourceV2Exec.scala, wrap / rethrow	QueryExecutionErrors.writingJobFailedError
v3.4.3	434 / 437	912
v3.5.8	416 / 419	899
v4.0.1	452 / 455	889
v4.1.1	483 / 486	944
v4.2.0	664 / 667	973

IcebergCommitExec already matched Spark when the abort succeeds. It diverged only when the abort failed: it attached the abort failure as suppressed but rethrew the raw cause instead of wrapping it. So the fix is narrower than the issue suggests, and needs no version shim.

What changes are included in this PR?

• IcebergCommitExec: on both the job-failure and the commit-failure path, when batchWrite.abort throws, throw writingJobFailedError(cause) instead of cause. A single helper, abortAfter, runs the abort and returns what the failed write throws: the cause itself when the abort succeeds, or the wrapped cause when it fails. Logging an abort failure logs only the message, as Spark does, since its stack trace travels with the thrown exception as a suppressed exception. Everything else is unchanged: the abort itself, the deletion of completed tasks' data files on a job failure, and attaching abort and cleanup failures to the cause as suppressed exceptions. Cleanup failures alone do not cause wrapping, because that cleanup is Comet's own addition and has no counterpart in Spark's path.
• CometIcebergWriteActionSuite, four tests:
  • a commit-time validation failure (a serializable overwrite validated against an older snapshot, which fails deterministically without concurrency), run with the split operator off and on, asserting that both failures carry the expected message, that the thrown exception and its cause have the same classes, and that the split run went through IcebergCommitExec
  • a failed write job (a UDF that fails a task over a Parquet source), with the same off/on comparison
  • an abort that fails after a commit failure, and one after a job failure, each driving IcebergCommitExec with a stub BatchWrite, asserting the writingJobFailedError wrapping, the cause, and the suppressed abort failure

One thing noticed while writing the job-failure test: the existing "failed write job aborts and leaves the table unchanged" test uses a local relation as its source, so on Spark 4.1 the optimizer evaluates its UDF during planning and the query fails before a write job starts. The new job-failure test reads from Parquet so that a task actually fails. The existing test is unchanged here.

How are these changes tested?

• Default profile (Spark 4.1): ./mvnw test -Dtest=none -Dsuites="org.apache.comet.CometIcebergWriteActionSuite", 73 succeeded, 0 failed.
• Spark 3.5: the same suite with -Pspark-3.5, 70 succeeded, 0 failed, 3 canceled (tests gated on Spark 4.1+).
• The two abort-failure tests fail against main's IcebergCommitExec.

[sunchao]

Summary

• Prior state and problem: IcebergCommitExec preserved the original write failure even when BatchWrite.abort failed. Supported Spark versions wrap that case in writingJobFailedError, with the abort failure suppressed on the original cause.
• Design approach: The small abortAfter helper shares that behavior between job and commit failures. It returns the throwable so completed-task cleanup still runs after a job failure.
• Correctness / compatibility analysis: Reviewed at 6d674be4f067644544cc8891fa0b03a4ba14d72f. The helper matches upstream Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. The wrapper retains the original cause, including cleanup errors suppressed after wrapping. I found no reproducible P1/P2 regression.
• Key design decisions: Abort success preserves the original exception object. Only abort failure adds Spark's wrapper. Cleanup failure alone keeps the prior behavior, and the successful write path is unchanged.
• Implementation sketch: Both failure handlers call abortAfter. It invokes abort, records a suppressed abort error when necessary, and returns either the original cause or QueryExecutionErrors.writingJobFailedError(cause). The tests cover task and commit failures with successful and failed aborts.
• Behavioral changes worth calling out: Callers now receive SparkException("Writing job failed.") when both the write and abort fail. The original failure remains its cause. No row-processing or successful-write overhead was added.
• Suggested improvements: No additional code changes identified at the requested P1/P2 bar.

Validation: the exact-head scans CI job passed all 73 CometIcebergWriteActionSuite tests, including all four new tests and the existing completed-task cleanup cases. The head has 24 passing and 14 skipped checks. I did not rerun the suite locally. Broader Iceberg SQL and non-default runtime coverage remains outside this PR-tier evidence, so this review does not establish that the repository's pre-queue integration validation is complete.