Preserve DSv2 metadata writes and reinserts for SPARK-50820

[firestarman]

Fixes #15348.

Description

• Route GPU ReplaceDataExec through a metadata-aware writing task when ReplaceDataProjections.metadataProjection is present, so CoW DSv2 writes call writer.write(metadata, row) like Spark's SPARK-50820 CPU path.
• Split GPU Delta insert and reinsert paths so REINSERT_OPERATION rows call DeltaWriter.reinsert(...) with metadata when available, instead of treating them as ordinary inserts.
• Add GpuDsv2WriteMetadata and Iceberg GpuRowLineage so GPU Iceberg writers can decorate batches with _row_id / _last_updated_sequence_number during metadata writes and reinserts.
• Add unit coverage in GpuDsv2MetadataWriteSuite and an Iceberg V3 CoW UPDATE integration test (test_iceberg_row_lineage_update_cow) that compares CPU/GPU user columns plus row-lineage metadata.

Validation:

• mvn -s /home/liangcail/.m2/settings_art.xml -f scala2.13/pom.xml -Dbuildver=400 -Dcuda.version=cuda13 -DskipTests validate
• mvn -s /home/liangcail/.m2/settings_art.xml -f scala2.13/pom.xml -Dbuildver=400 -Dcuda.version=cuda13 -pl tests test -DwildcardSuites=org.apache.spark.sql.execution.datasources.v2.GpuDsv2MetadataWriteSuite
• Spark 4.1.1 + Iceberg 1.11.0: test_iceberg_row_lineage_update_cow passed with CPU Iceberg reads and GPU Iceberg writes

Notes:

• Integration test forces CPU Iceberg reads because GPU Iceberg scans do not yet inherit V3 row-lineage columns.
• MoR V3 coverage is deferred: Iceberg V3 MoR uses PUFFIN deletion vectors, while GPU WriteDelta is Parquet-only today.

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
  (Please provide the names of the existing tests in the PR description.)
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Greptile Summary

Preserve Spark 4 DSv2 metadata writes and Iceberg row lineage during copy-on-write and reinsert operations.

• Separates INSERT and REINSERT operation handling in Spark-version shims.
• Adds task-local metadata-schema propagation and metadata-aware ReplaceData routing.
• Decorates Iceberg write batches with row-lineage columns.
• Adds Spark 4 unit coverage and an Iceberg V3 copy-on-write integration test.

Confidence Score: 4/5

The metadata-aware ReplaceData path needs OOM retry protection before merging because recoverable GPU memory pressure currently fails the write.

The new task allocates projected and filtered GPU batches outside the repository’s retry framework, unlike the analogous delta-writing path, so a retryable RMM allocation failure aborts the operation.

Files Needing Attention: sql-plugin/src/main/spark400/scala/org/apache/spark/sql/execution/datasources/v2/GpuReplaceDataExec.scala

Important Files Changed

Filename	Overview
sql-plugin/src/main/spark400/scala/org/apache/spark/sql/execution/datasources/v2/GpuReplaceDataExec.scala	Adds metadata-aware ReplaceData routing, but its new GPU projection and filtering path lacks OOM retry handling.
sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/DeltaInsertFilter.scala	Separates Spark 4 INSERT and REINSERT rows and propagates metadata schema around reinsert calls.
sql-plugin/src/main/spark350/scala/com/nvidia/spark/rapids/shims/DeltaInsertFilter.scala	Preserves Spark 3.5 behavior through compatible insert helpers and a no-op reinsert shim.
iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuRowLineage.scala	Adds lineage-column decoration with explicit ownership handling for data and metadata batches.
iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkWrite.scala	Makes partitioned and unpartitioned Iceberg writers metadata-aware and decorates lineage-enabled writes.
iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkPositionDeltaWrite.scala	Routes inserts through reinsert semantics and decorates position-delta data with lineage metadata.
integration_tests/src/main/python/iceberg/iceberg_row_lineage_test.py	Compares CPU and GPU Iceberg V3 copy-on-write updates, including row-lineage metadata.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[ReplaceData batch] --> B{Operation}
  B -->|WRITE| C[writer.write data]
  B -->|WRITE_WITH_METADATA| D[Project metadata and data]
  D --> E[writer.write metadata, data]
  E --> F[GpuRowLineage.decorate]
  F --> G[Iceberg data writer]
  H[Delta write batch] --> I{INSERT or REINSERT}
  I -->|INSERT| J[writer.insert]
  I -->|REINSERT| K[writer.reinsert metadata, row]
  K --> F

Loading

_{Reviews (1): Last reviewed commit: "Add Iceberg V3 row-lineage CoW UPDATE in..." | Re-trigger Greptile}

[greptile-apps]

Metadata writes bypass OOM retry

When a metadata-aware copy-on-write batch encounters a retryable GPU allocation failure during projection, filtering, or lineage decoration, this task performs the allocation outside withRetry or withRetryNoSplit, causing the write operation to abort instead of spilling and retrying.

Rule Used: GPU memory-allocating code must be wrapped in with... (source)