test: cover native Iceberg mid-write retry cleanup

[sam-1112]

Which issue does this PR close?

Closes part of #5646 (native mid-write failure, retry, and failed-attempt file cleanup).

Rationale for this change

Add coverage for a native Iceberg writer failure after data files have already been finalized.

The test injects a file:// write failure during a native write, verifies that Spark retries the
failed task, and confirms that the failed attempt leaves no committed or orphaned data files.

What changes are included in this PR?

• Add a native Iceberg mid-write failure and retry test.
• Use local[5,2] so the local Spark master allows one task retry, with a comment documenting the suite-wide setting.
• Force the native writer to fail with EISDIR after earlier data files have been finalized.
• Arm four consecutive blocker paths (max + 1 through max + 4) to tolerate timing differences
  between the projection and native writer pipeline.
• Remove all blocker directories during the retry and verify their cleanup.
• Preserve the first-attempt probe failure reason so CI failures report the underlying cause.
• Share the Parquet directory-walking logic between the suite and retry probe.
• Count only regular .parquet files, excluding test-created blocker directories.
• Verify the retry produces exactly one Iceberg snapshot.
• Verify physical data files exactly match the files referenced by Iceberg metadata.
• Verify files from the failed attempt are neither present in storage nor referenced by manifests.
• Verify the final table contains all expected rows without duplicates.

How are these changes tested?

• The Spark 3.5 retry test was run 20 consecutive times:

  for i in $(seq 1 20); do
    ./mvnw test -Pspark-3.5 -Dtest=none \
      -Dsuites="org.apache.comet.CometIcebergWriteActionSuite mid-write failure retries" \
      || break
  done

Result: 20/20 runs passed.

• Full CometIcebergWriteActionSuite with Spark 3.5:

./mvnw test -Pspark-3.5 -Dtest=none \
    -Dsuites="org.apache.comet.CometIcebergWriteActionSuite"

Result: 67 succeeded, 0 failed, 3 canceled.

• Full CometIcebergWriteActionSuite with the default Spark 4.1 profile:

./mvnw test -Dtest=none \
    -Dsuites="org.apache.comet.CometIcebergWriteActionSuite"

Result: 70 succeeded, 0 failed, 0 canceled.

[andygrove]

Thanks for this. The gap is a real one and the test is well constructed. physical == referenced paired with failedPaths.nonEmpty is the right central shape, it cannot pass vacuously, and the count(*) versus count(DISTINCT id) pair covers the duplicate rows a bad retry would leave behind. The comments explaining the EISDIR trick and the attempt-unique naming match how the rest of this suite documents itself.

A few things I would like to settle before merge.

setMaster("local[5,2]") is suite-wide

This is doing more than it looks like. local[N] pins maxFailures to 1 and local[N,2] raises it to 2, and spark.task.maxFailures is ignored under a local master, so I agree the master string is the only lever. But it applies to every test in the suite, not just the new one.

Two consequences. The six existing failure-injection tests now run their injected failure twice: failed write job aborts and leaves the table unchanged, a failed task deletes the data files it already finalized, and both native= variants of a failed write job deletes the data files of tasks that completed. I walked JobAbortGate and ConflictGate and I do not think anything deadlocks. awaitOthers() on the retry returns immediately because the latch is already at zero, and finished >= 2 still holds. But those abort and cleanup paths now have to be correct across two attempts rather than one, which is a different assertion than the one they were written for.

The second consequence worries me more. This suite is what pins write, abort, and cleanup semantics, and it will now silently retry any one-off failure anywhere in it rather than reporting it.

The testing section only shows the new test run by name, which does not exercise the change to the other seventy. Could you run the full CometIcebergWriteActionSuite and paste the result into the description? A comment on the setMaster line explaining why the ,2 is there would help too. The rest of this suite explains its setup carefully and that detail is easy to delete by accident later.

The blocker file number is derived from timing

In NativeWriteRetryProbe.check the blocked path is max(finalized file numbers) + 1, measured at the instant row 7000 is projected. That is the writer's next file only if its internal counter is exactly in step with what has reached disk right then, and there is FFI pipelining between the projection and the writer. One file ahead and createDirectory throws on a path that already exists. One file behind and nothing is ever blocked, so no retry happens.

Both skews fail the test rather than passing it falsely, which is the right direction. But what pins the counter to the on-disk state here? Would it be worth arming a small range, say max+1 through max+4, so a little skew does not disarm the injection? Either way, could you run this one about twenty times and report how it holds up? Iceberg flakes are painful to chase down later.

Attempt-0 diagnostics get swallowed

The three IllegalStateExceptions in check (native writer did not finalize before injection, unexpected native file name, expected one native task prefix) all throw on attempt 0. Spark treats that as a normal task failure, so the retry runs, the write succeeds, and the test fails on assert(blockerCreated, "the storage failure was not armed") with no trace of which check actually tripped. The real message only reaches the executor log. Could the probe stash the reason in a field and have that assertion include it? That would make a CI failure readable without digging through logs.

parquetFiles matches directories

Neither parquetFiles nor the copy of that walk inside check filters to regular files, and the blocker is a directory named ...-000NN.parquet sitting in the table's data/ directory. It is only removed inside the retry's UDF call at row 7000, so on any path that does not reach that row the directory survives and gets counted as a physical data file. The test then reports orphan files: <the blocker>, which points the reader at cleanup when cleanup did nothing wrong. Adding a Files.isRegularFile check to the filter would sort that out.

The probe's walk is also a copy of parquetFiles. I understand the probe has to stay top-level so the closure does not capture the suite. Could the walk move into a small shared object so there is only one copy to fix?

[sam-1112]

Thanks @andygrove . I pushed an update for the review comments.

• Added a note for why this suite uses local[5,2].
• Changed the blocker setup to cover a few candidate output paths instead of relying on only max + 1.
• Kept the first-attempt failure around so the final assertion shows the actual reason when the retry setup doesn't behave as expected.
• Moved the parquet file scan into IcebergTestFiles and filtered it to regular files only.

I also reran the suite and repeated the new retry test to check for flakiness.