[SPARK-56743][SQL] Use SQLLastAttemptMetric for DSv2 UPDATE/DELETE/MERGE metrics

[juliuszsompolski]

What changes were proposed in this pull request?

Switches the DSv2 row-level operation metrics that count rows produced by executor tasks to use SQLLastAttemptMetric instead of plain SQLMetric, so the values surfaced in UpdateSummary, DeleteSummary, and MergeSummary are stable across stage retries.

Specifically:

• In RowLevelWriteExec.sparkMetrics, the UPDATE and DELETE branches now construct their row counters (numUpdatedRows / numDeletedRows / numCopiedRows) via SQLLastAttemptMetrics.createMetric. Increment paths are unchanged (SQLLastAttemptMetric extends SQLMetric).
• In MergeRowsExec.metrics, all 8 row counters (numTargetRowsCopied, numTargetRowsInserted, numTargetRowsDeleted, numTargetRowsUpdated, and the matched / not-matched-by-source splits) are switched to SLAM. Both the interpreted (longMetric("...") += 1) and codegen (metricTerm(...).add(1)) increment paths work unchanged.
• BatchScanExec overrides sparkMetrics so numOutputRows becomes SLAM only when the scan reads on behalf of a row-level DELETE (i.e. table is RowLevelOperationTable with operation.command() == DELETE). This is needed because group-based DELETE derives numDeletedRows = numScannedRows - numCopiedRows on the driver in ReplaceDataExec.getWriteSummary; if either input is overcounted the difference is overcounted in lockstep on a writer-stage retry. Other scans (regular reads, UPDATE/MERGE row-level scans, streaming scans via the trait default) keep the plain SQLMetric.
• RowLevelWriteExec.getMetricValue now reads via lastAttemptValueForHighestRDDId() for SQLLastAttemptMetric, falling back to slam.value if SLAM bailed out. This handles the UPDATE/DELETE summary paths, the MERGE summary path (MergeRowsExec.metrics), and the scan side numOutputRows referenced from ReplaceDataExec.getWriteSummary uniformly.

The driver-side metrics("numDeletedRows").set(...) in ReplaceDataExec.getWriteSummary continues to work after the swap because for the group-based DELETE path that metric is only updated on the driver, so SLAM records it as a driver value and surfaces it via lastAttemptValueForHighestRDDId() without bailing out.

Why are the changes needed?

SQLMetric.value aggregates increments from every task attempt that ever ran, so on a stage retry the row counts double up. The values flow into the connector-visible WriteSummary (and downstream into operator metrics consumers such as Delta's invariant checks), so an inflated count mis-reports what the operation actually did. SQLLastAttemptMetric reports only the last attempt's contribution and so gives the row count that matches what was actually committed.

Does this PR introduce any user-facing change?

The values surfaced via UpdateSummary.numUpdatedRows / numCopiedRows, DeleteSummary.numDeletedRows / numCopiedRows, and MergeSummary.numTargetRows* (i.e. what the connector receives from BatchWrite.commit(messages, summary)) will, in the presence of stage retries, be the row counts from the last attempt rather than the sum across all attempts. With no retries, behavior is unchanged. The metric names, display strings, and presence in the SQL UI are unchanged.

Note: the SQL UI still shows the raw accumulator value (SQLMetric.value), which on stage retries is the sum across all task attempts and therefore overcounts. Only the values passed to the connector via WriteSummary are SLAM-corrected. Making the SQL UI also display the last-attempt value is a bigger follow-up that would touch the SQL UI's metric collection pipeline.

How was this patch tested?

Existing tests in UpdateTableSuiteBase, DeleteFromTableSuiteBase, and the merge suites exercise the metric values; they continue to pass since SLAM and SQLMetric report the same value when there are no stage retries.

Three new tests cover the retry behavior directly, one in each of UpdateTableSuiteBase, DeleteFromTableSuiteBase, and MergeIntoTableSuiteBase, all named "metric values are stable across stage retries". Each runs the operation with an IN-subquery (or a join, for MERGE) to force a shuffle, sets spark.sql.autoBroadcastJoinThreshold = -1 so the join doesn't get broadcast away (AQE is left at its default), and then enables spark.test.injectShuffleFetchFailures. The DAGScheduler corrupts the first attempt of each shuffle map stage, the writer stage retries, and the test asserts that the corresponding *Summary row counts still match the actual operation result. Each test runs across the suite's variants (group-based / delta-based, with/without metadata).

The injected retries are visible in the test logs as FetchFailedException / MetadataFetchFailedException. The summary-value assertions only pass because the new SLAM-aware reader returns the last-attempt value rather than the doubled raw accumulator value.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 4.7

[juliuszsompolski]

@ZiyaZa @aokolnychyi

[ZiyaZa]

Any reason for not always using SLAM here? Is it riskier than regular metrics?

IMO it would look better if this code was in DataSourceV2ScanExecBase, but there we can't check if it's a DELETE command.

[juliuszsompolski]

There is some memory overhead - an array of num tasks size of partial values by default, so let's say a couple hundred bytes to a couple of kilobytes. Computational overhead should be negligible.
I think that in general all numOutputRows metrics all over Spark could benefit from porting to SLAM, but wanted to make the blast radius as small as possible here for now.

[juliuszsompolski]

TODO: Work on a better test right not for UPDATE and MERGE, because the fetch failure injection framework right now doesn't trigger retries of the last stage, where these metrics are computed, so actually doesn't trigger duplicate non SLAM metrics.
Such duplicates would happen if the downstream stage produces a checksum mismatch on recompute, so the final stage gets restarted with a full retry. The failure injection mechanism would need some extensions to simulate that.