Pass DSv2 WriteSummary from GPU MERGE commits [databricks] - #15429
Pass DSv2 WriteSummary from GPU MERGE commits [databricks]#15429firestarman wants to merge 6 commits into
Conversation
Mirror SPARK-52689/SPARK-53891 so Spark 4.1+ BatchWrite.commit receives MergeSummary metrics collected from GpuMergeRowsExec, and Iceberg GPU BatchWrite wrappers forward the summary instead of dropping it. Signed-off-by: Firestarman <firestarmanllc@gmail.com>
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Greptile SummaryThe PR forwards Spark 4.1+ DSv2 MERGE write summaries through GPU write paths.
Confidence Score: 5/5The PR appears safe to merge. No blocking failure remains. Important Files Changed
Sequence DiagramsequenceDiagram
participant Merge as GpuMergeRowsExec
participant Retry as withRetryNoSplit
participant Metrics as Attempt Metrics
participant Write as GpuV2TableWriteExec
participant Batch as BatchWrite
participant Delegate as CPU Delegate
Merge->>Retry: Process input batch
loop Each retry attempt
Retry->>Metrics: Reset staging counters
Retry->>Merge: Apply MERGE instructions
Merge->>Metrics: Record row actions
end
Retry-->>Merge: Successful output batch
Merge->>Metrics: Publish successful counters
Write->>Batch: commit(messages, MergeSummary)
Batch->>Delegate: Forward summary commit
Reviews (6): Last reviewed commit: "Fix CI: iceberg summary suite on 411 onl..." | Re-trigger Greptile |
Move org.scalatest into the 3rdParty import group ahead of org.apache.spark so verify's all-module scalastyle check passes. Signed-off-by: Firestarman <firestarmanllc@gmail.com>
Record GpuMergeRowsExec WriteSummary counters into LocalGpuMetric staging inside withRetryNoSplit and flush only after a successful attempt so a retryable GPU OOM cannot double-count MERGE rows. Signed-off-by: Firestarman <firestarmanllc@gmail.com>
|
build |
| var attemptMetrics: MergeRowMetrics = null | ||
| val result = withRetryNoSplit(batch) { _ => | ||
| // Fresh staging metrics for each attempt so failed retries are discarded. | ||
| attemptMetrics = MergeRowMetrics.local() |
There was a problem hiding this comment.
[SUGGESTION] Reuse per-iterator staging metrics
This retry callback allocates a new MergeRowMetrics and eight LocalGpuMetric objects for every batch attempt. A GpuMergeBatchIterator processes batches serially, so the staging object can be reused without changing retry semantics as long as it is cleared at the start of every attempt.
Expected change: Allocate one staging MergeRowMetrics per GpuMergeBatchIterator, add a reset method that sets all eight mutable metrics to zero, and call that reset at the beginning of each withRetryNoSplit attempt before recording any rows.
There was a problem hiding this comment.
Done in 973a167: GpuMergeBatchIterator now owns one reused attemptMetrics and calls reset() at the start of every withRetryNoSplit attempt.
| * Without this override, the default interface method calls `commit(messages)` and drops the | ||
| * summary, so Iceberg/Delta commit metadata would miss DML metrics on the GPU path. | ||
| */ | ||
| trait GpuV2BatchWriteSummaryCommit { this: BatchWrite => |
There was a problem hiding this comment.
[SHOULD FIX] Disable unused MERGE summary accounting before Spark 4.1
The shared GpuMergeRowsExec path now creates eight staging metrics per batch/retry attempt and publishes eight counters after success on Spark 3.5 through 4.2. Before Spark 4.1, GpuV2WriteCommitShims always calls the one-argument commit, so those counters cannot reach a WriteSummary; the allocation, classification, and metric-update work is discarded.
Expected change: Add a compile-time shim capability flag that is false before Spark 4.1 and true on Spark 4.1+, use a singleton MergeRowMetrics backed by NoopMetric when the capability is disabled, and skip publishing disabled staging metrics. Preserve the current retry-safe accounting on Spark 4.1+.
There was a problem hiding this comment.
Done in 973a167: added GpuMergeRowMetricsShims.writeSummaryEnabled (false before 4.1 / true on 4.1+). Pre-4.1 skips MergeSummary metric registration and uses MergeRowMetrics.noop (NoopMetric).
There was a problem hiding this comment.
Additional gaps called out in re-verification are addressed in 8b96b00 (NOOP rename, skip reset/record/addAll pre-4.1, forAttempt coverage, descendant traversal, ESSENTIAL session, real GpuBatchAppend/GpuPositionDeltaBatchWrite).
|
|
||
| /** Visible for tests. */ | ||
| private[v2] def getWriteSummary(query: SparkPlan): Option[WriteSummary] = { | ||
| collectFirst(query) { case m: GpuMergeRowsExec => m }.map { mergeRows => |
There was a problem hiding this comment.
[SHOULD FIX] Preserve summaries for CPU MergeRowsExec fallback plans
getWriteSummary currently matches only GpuMergeRowsExec. The V2 writer can still be on GPU while its MERGE child falls back to CPU, and Spark's native V2 path collects the CPU MergeRowsExec metrics in that case. The current match returns None for this supported mixed plan and drops the MERGE summary by selecting the one-argument commit.
Expected change: Collect the metrics map from either plan type and pass it to mergeSummaryFromMetrics:
collectFirst(query) {
case m: GpuMergeRowsExec => m.metrics
case m: MergeRowsExec => m.metrics
}.map(mergeSummaryFromMetrics)Add coverage for the CPU-child/GPU-writer fallback so both plan types select the summary-aware commit overload.
There was a problem hiding this comment.
Done in 973a167: getWriteSummary now matches GpuMergeRowsExec | MergeRowsExec and feeds either metrics map into mergeSummaryFromMetrics. Covered by "commit preserves summary for CPU MergeRowsExec fallback child".
| assert(m.numTargetRowsDeleted.value === 0) | ||
| } | ||
|
|
||
| test("addAll flushes a successful attempt and discards failed-attempt staging") { |
There was a problem hiding this comment.
[SHOULD FIX] Exercise the production retry path in this regression test
This test manually creates failed and successful staging objects and calls addAll only for the successful one. It never executes GpuMergeBatchIterator, withRetryNoSplit, or an injected OOM, so moving publication back inside the retry callback or reusing failed-attempt metrics would still leave the test green.
Expected change: Replace the manual simulation with a test based on RmmSparkRetrySuiteBase that builds a one-row GpuMergeBatchIterator, injects a retryable GPU OOM with RmmSpark.forceRetryOOM after all inputs are built, executes iterator.next(), verifies that a retry was observed, and asserts that every published MERGE counter contains only the successful attempt. The test must own the required RMM setup and cleanup.
There was a problem hiding this comment.
Done in 973a167: added GpuMergeBatchIteratorRetrySuite (tests module, RmmSparkRetrySuiteBase + forceRetryOOM after record / during project) so the real withRetryNoSplit publication boundary is exercised. The unit suite keeps a lighter reset/addAll check.
| assert(recorder.plainCommitCount === 1) | ||
| } | ||
|
|
||
| test("mergeSummaryFromMetrics reads GpuMergeRowsExec metric names") { |
There was a problem hiding this comment.
[MUST FIX] Test the public summary-aware commit path end to end
The suite calls mergeSummaryFromMetrics directly, but no test invokes GpuV2WriteCommitShims.commit or exercises its GpuMergeRowsExec plan traversal. It also leaves five summary fields unchecked. Returning None from traversal, selecting the plain commit overload, or swapping unchecked constructor fields would therefore pass.
Expected change: Build a real GpuMergeRowsExec plan, seed all eight metrics with distinct values, invoke GpuV2WriteCommitShims.commit with a recording BatchWrite, and assert all eight MergeSummary fields received by the summary-aware overload. Also assert that the summary commit is called exactly once and the plain commit is not called.
There was a problem hiding this comment.
Done in 973a167: GpuV2WriteSummarySuite now builds a real GpuMergeRowsExec, seeds all eight metrics, calls GpuV2WriteCommitShims.commit, and asserts summary-aware overload + all eight MergeSummary fields.
There was a problem hiding this comment.
Additional gaps called out in re-verification are addressed in 8b96b00 (NOOP rename, skip reset/record/addAll pre-4.1, forAttempt coverage, descendant traversal, ESSENTIAL session, real GpuBatchAppend/GpuPositionDeltaBatchWrite).
| SQLMetrics.createMetric(sc, "nmbsUpdated"), | ||
| GpuMergeRowsExec.NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_DELETED -> | ||
| SQLMetrics.createMetric(sc, "nmbsDeleted")) | ||
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_COPIED).add(10) |
There was a problem hiding this comment.
[SHOULD FIX] Cover every metric mapping and the missing-key sentinel
This test gives non-zero values only to copied and inserted and asserts only copied, inserted, and a zero-valued deleted field. Swapped or defaulted values for the other five fields still pass, and the -1L missing-key fallback is never exercised because every key is present.
Expected change: Assign a distinct non-zero value to each of the eight input metrics and assert every corresponding MergeSummaryImpl field. Then remove one metric key and assert that its output field is -1L, covering the missing-value path.
There was a problem hiding this comment.
Done in 973a167: mergeSummaryFromMetrics test now assigns distinct non-zero values to all eight fields and asserts each mapping, then drops one key and asserts the -1L sentinel.
| NUM_TARGET_ROWS_MATCHED_UPDATED -> createMetric(MODERATE_LEVEL, | ||
| "number of target rows updated by a matched clause"), | ||
| NUM_TARGET_ROWS_MATCHED_DELETED -> createMetric(MODERATE_LEVEL, | ||
| "number of target rows deleted by a matched clause"), | ||
| NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_UPDATED -> createMetric(MODERATE_LEVEL, | ||
| "number of target rows updated by a not matched by source clause"), | ||
| NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_DELETED -> createMetric(MODERATE_LEVEL, | ||
| "number of target rows deleted by a not matched by source clause") |
There was a problem hiding this comment.
MUST FIX — Keep committed MERGE summary fields available at ESSENTIAL
These four per-clause counts are commit metadata, but MODERATE_LEVEL turns them into NoopMetric when spark.rapids.sql.metrics.level=ESSENTIAL. They are then absent from SparkPlan.metrics, so mergeSummaryFromMetrics commits -1 for rows that were actually updated or deleted. The metrics level must not change the table metadata produced by MERGE.
Expected code change: register all four fields consumed by MergeSummaryImpl at ESSENTIAL_LEVEL.
| NUM_TARGET_ROWS_MATCHED_UPDATED -> createMetric(MODERATE_LEVEL, | |
| "number of target rows updated by a matched clause"), | |
| NUM_TARGET_ROWS_MATCHED_DELETED -> createMetric(MODERATE_LEVEL, | |
| "number of target rows deleted by a matched clause"), | |
| NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_UPDATED -> createMetric(MODERATE_LEVEL, | |
| "number of target rows updated by a not matched by source clause"), | |
| NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_DELETED -> createMetric(MODERATE_LEVEL, | |
| "number of target rows deleted by a not matched by source clause") | |
| NUM_TARGET_ROWS_MATCHED_UPDATED -> createMetric(ESSENTIAL_LEVEL, | |
| "number of target rows updated by a matched clause"), | |
| NUM_TARGET_ROWS_MATCHED_DELETED -> createMetric(ESSENTIAL_LEVEL, | |
| "number of target rows deleted by a matched clause"), | |
| NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_UPDATED -> createMetric(ESSENTIAL_LEVEL, | |
| "number of target rows updated by a not matched by source clause"), | |
| NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_DELETED -> createMetric(ESSENTIAL_LEVEL, | |
| "number of target rows deleted by a not matched by source clause") |
There was a problem hiding this comment.
Done in 973a167: all eight MergeSummary metric fields are registered at ESSENTIAL_LEVEL so metrics.level cannot turn commit metadata into NoopMetric/-1.
There was a problem hiding this comment.
Additional gaps called out in re-verification are addressed in 8b96b00 (NOOP rename, skip reset/record/addAll pre-4.1, forAttempt coverage, descendant traversal, ESSENTIAL session, real GpuBatchAppend/GpuPositionDeltaBatchWrite).
| spillable.get.getColumnarBatch() | ||
| } | ||
| } | ||
| mergeMetrics.addAll(attemptMetrics) |
There was a problem hiding this comment.
SHOULD FIX — Exercise the real retry boundary that protects metric publication
The current test creates separate failed and successful MergeRowMetrics instances and explicitly publishes only the successful one. It never constructs GpuMergeBatchIterator, enters withRetryNoSplit, or injects an OOM, so moving this addAll call into the retry callback would reintroduce double counting without failing the test.
Expected test change: add a GPU test in the tests module using RmmSparkRetrySuiteBase that constructs a real one-batch GpuMergeBatchIterator, uses RmmSpark.forceRetryOOM to fail an attempt after row counts have been recorded but before the retry callback returns, consumes iterator.next(), verifies RmmSpark.getAndResetNumRetryThrow(/* taskId = */ 1) > 0, and asserts every affected published count equals one successful execution rather than failed plus successful attempts. Keep the injection offset tied to the verified allocation immediately after attemptMetrics.record so the test fails if publication moves back inside the callback.
There was a problem hiding this comment.
Done in 973a167: added tests/.../GpuMergeBatchIteratorRetrySuite which builds a real GpuMergeBatchIterator, injects forceRetryOOM after attemptMetrics.record (during applyOutputs/project), verifies a retry occurred, and asserts published counts equal one successful attempt.
| collectFirst(query) { case m: GpuMergeRowsExec => m }.map { mergeRows => | ||
| mergeSummaryFromMetrics(mergeRows.metrics) | ||
| } |
There was a problem hiding this comment.
SHOULD FIX — Preserve MERGE summaries for mixed CPU/GPU plans
This matcher recognizes only GpuMergeRowsExec. The outer Iceberg write can remain on GPU while an unsupported MergeRowsExec child falls back to CPU and is wrapped by GpuRowToColumnarExec; in that supported mixed plan, this returns None and the one-argument commit drops the valid CPU MERGE summary. Both nodes expose the same eight Spark metric names.
Expected code change: collect the metrics map from either node and pass it to the existing helper.
| collectFirst(query) { case m: GpuMergeRowsExec => m }.map { mergeRows => | |
| mergeSummaryFromMetrics(mergeRows.metrics) | |
| } | |
| collectFirst(query) { | |
| case m: GpuMergeRowsExec => m.metrics | |
| case m: MergeRowsExec => m.metrics | |
| }.map(mergeSummaryFromMetrics) |
There was a problem hiding this comment.
Done in 973a167: getWriteSummary now collects metrics from GpuMergeRowsExec | MergeRowsExec and passes them to mergeSummaryFromMetrics. Covered by the CPU MergeRowsExec fallback commit test in GpuV2WriteSummarySuite.
There was a problem hiding this comment.
Additional gaps called out in re-verification are addressed in 8b96b00 (NOOP rename, skip reset/record/addAll pre-4.1, forAttempt coverage, descendant traversal, ESSENTIAL session, real GpuBatchAppend/GpuPositionDeltaBatchWrite).
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_COPIED).add(10) | ||
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_INSERTED).add(20) | ||
| val summary = GpuV2WriteCommitShims.mergeSummaryFromMetrics(metrics) | ||
| assert(summary.numTargetRowsCopied === 10) | ||
| assert(summary.numTargetRowsInserted === 20) | ||
| assert(summary.numTargetRowsDeleted === 0) |
There was a problem hiding this comment.
SHOULD FIX — Populate and assert all eight summary fields with distinct values
Only copied and inserted receive non-zero values, and only copied, inserted, and deleted are asserted. Because MergeSummaryImpl is constructed positionally from eight named metrics, the unchecked zero-valued fields can be permuted or omitted without this test detecting it.
Expected code change: replace this partial setup and assertion block with complete independent coverage.
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_COPIED).add(10) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_INSERTED).add(20) | |
| val summary = GpuV2WriteCommitShims.mergeSummaryFromMetrics(metrics) | |
| assert(summary.numTargetRowsCopied === 10) | |
| assert(summary.numTargetRowsInserted === 20) | |
| assert(summary.numTargetRowsDeleted === 0) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_COPIED).add(1L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_DELETED).add(2L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_UPDATED).add(3L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_INSERTED).add(4L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_MATCHED_UPDATED).add(5L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_MATCHED_DELETED).add(6L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_UPDATED).add(7L) | |
| metrics(GpuMergeRowsExec.NUM_TARGET_ROWS_NOT_MATCHED_BY_SOURCE_DELETED).add(8L) | |
| val summary = GpuV2WriteCommitShims.mergeSummaryFromMetrics(metrics) | |
| assert(summary.numTargetRowsCopied === 1L) | |
| assert(summary.numTargetRowsDeleted === 2L) | |
| assert(summary.numTargetRowsUpdated === 3L) | |
| assert(summary.numTargetRowsInserted === 4L) | |
| assert(summary.numTargetRowsMatchedUpdated === 5L) | |
| assert(summary.numTargetRowsMatchedDeleted === 6L) | |
| assert(summary.numTargetRowsNotMatchedBySourceUpdated === 7L) | |
| assert(summary.numTargetRowsNotMatchedBySourceDeleted === 8L) |
There was a problem hiding this comment.
Done in 973a167: mergeSummaryFromMetrics now seeds all eight metrics with distinct non-zero values and asserts every MergeSummaryImpl field, then drops one key and asserts the -1L missing-key sentinel. The e2e commit tests similarly seed/assert all eight fields.
…rage Reuse per-iterator staging metrics with reset, disable MERGE summary accounting before Spark 4.1 via NoopMetric shims, collect CPU MergeRowsExec fallback metrics, keep all MergeSummary fields at ESSENTIAL_LEVEL, and add end-to-end commit plus RMM retry regression tests for the publication boundary. Signed-off-by: Firestarman <firestarmanllc@gmail.com>
|
build |
Rename MergeRowMetrics.NOOP, skip reset/record/addAll before Spark 4.1, exercise forAttempt capability and descendant plan traversal under ESSENTIAL metrics level, and cover real GpuBatchAppend / GpuPositionDeltaBatchWrite summary forwarding. Signed-off-by: Firestarman <firestarmanllc@gmail.com>
|
Follow-up in 8b96b00 addressing the remaining verification gaps against head 973a167:
|
Restrict GpuIcebergWriteSummaryCommitSuite to Spark 4.1.1 where the real iceberg module is built; 412+ use iceberg-stub and lack GpuBatchAppend. Also eta-expand verifyParquetMagic for Scala 2.12 (same as NVIDIA#15476 / NVIDIA#15475). Signed-off-by: Firestarman <firestarmanllc@gmail.com>
|
CI follow-up in 8817d65:
|
|
build |
4 participants
Fixes #15349.
Description
SPARK-52689/SPARK-53891) expects DSv2 writes to pass aWriteSummaryintoBatchWrite.commit(messages, summary), but the GPU V2 write path still called single-argcommit(messages), so MERGE row metrics never reached Iceberg/Delta committers.GpuMergeRowsExec(copied/inserted/deleted/updated and matched / not-matched-by-source variants), buildMergeSummaryImplon Spark 4.1+, and commit through versionedGpuV2WriteCommitShims(no-op summary path before 4.1).LocalGpuMetricperwithRetryNoSplitattempt and flush only after success, so a retryable GPU OOM cannot double-count rows in the committedWriteSummary.GpuV2BatchWriteSummaryCommitinto Iceberg GPUBatchWritewrappers so the 2-argcommitforwards the summary to the CPU delegate instead of the default interface method dropping it.Keep.context→ action tags viaGpuMergeRowsKeepShimsso metric attribution matches Spark CPU on 4.1+.mvn -s ~/.m2/settings_art.xml -Dbuildver=330 -Dcuda.version=cuda13 -DskipTests validateSUCCESS;mvn -s ~/.m2/settings_art.xml -f scala2.13/pom.xml -Dbuildver=400 -Dcuda.version=cuda13 -DskipTests validateSUCCESS;mvn -s ~/.m2/settings_art.xml -f scala2.13/pom.xml -Dbuildver=411 -Dcuda.version=cuda13 -DskipTests validateSUCCESS; all-module scalastyle 0 errors; unit suitesGpuMergeRowMetricsSuite(5) andGpuV2WriteSummarySuite(5) passed on buildver=411.Known gaps (not in this PR): Append/Update/Delete write summaries, Iceberg IT for snapshot summary.
Checklists
Documentation
Testing
(Please provide the names of the existing tests in the PR description.)
Performance