Support Iceberg nested and binary GPU writes

[res-life]

Fixes #14239.

Depends on:

• Preserve LIST element field ids in Parquet output rapidsai/cudf#22143
• Expose getters for ColumnWriterOptions isBinary and Parquet field id rapidsai/cudf#22188
• [JAVA] ColumnWriterOptions.withBinary(name, nullable, parquetFieldId) places the field id on the wrong column; no public way to set parquetFieldId/isBinary on a list rapidsai/cudf#22347

Description

• Preserve Iceberg field ids and the original write schema so nested arrays, maps, and binary columns keep the Parquet layout and metrics Iceberg expects.
• Extend Iceberg integration coverage for append, CTAS, RTAS, and overwrite flows, including focused regression coverage for binary and array writes.
• Update ITs

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

This PR enables GPU-accelerated Iceberg writes for nested types (arrays, maps, structs) and binary columns by propagating Iceberg field IDs through the Spark schema metadata and into cuDF's ColumnWriterOptions. The core mechanism uses reflection to set private cuDF fields (hasParquetFieldId, parquetFieldId, isBinary) as a documented interim workaround until rapidsai/cudf#22347 lands public setters. Integration tests are substantially strengthened — the previous "expected fallback" tests are replaced with GPU-correctness checks using pre-materialized Parquet sources and assert_no_cpu_project_exec, and new focused binary/array-of-binary CTAS tests are added. The previous P1 concern (accidental removal of Hex support) is not present in this version; expr[Hex] is in GpuOverrides.scala.

Confidence Score: 5/5

Safe to merge; no P0/P1 issues found — the two comments are P2 style suggestions about a deliberately temporary reflection workaround.

The previous P1 (accidental Hex removal) is absent in this revision. No resource leaks, no broken field-ID round-trip logic, and no correctness gaps were found. The reflection-based cuDF field mutation is explicitly acknowledged with a TODO + tracking issue (rapidsai/cudf#22347). Test coverage is substantively improved. Score stays at the P2-only ceiling of 5.

sql-plugin/src/main/scala/com/nvidia/spark/rapids/SchemaUtils.scala — contains the reflection layer; should be revisited once cuDF#22347 ships public setters.

Important Files Changed

Filename	Overview
sql-plugin/src/main/scala/com/nvidia/spark/rapids/SchemaUtils.scala	Introduces reflection-based mutation of private cuDF ColumnWriterOptions fields (hasParquetFieldId, parquetFieldId, isBinary) as a workaround for missing public setters; adds nested field-ID propagation for ArrayType/MapType and rewrites BinaryType as a marked list; flagged with TODO + cuDF tracking issue.
iceberg/common/src/main/scala/org/apache/iceberg/spark/GpuTypeToSparkType.scala	Adds appendNestedFieldIdMetadata and nestedMetadataJson to encode Iceberg nested field IDs into Spark StructField metadata; iterates ListType/MapType children correctly via fields().asScala.
iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkFileWriterFactory.scala	Converts the single lazy TaskAttemptContext to a per-call newTaskAttemptContext(sparkType) that stamps ParquetWriteSupport.setSchema into the Hadoop conf; both data and delete writers now get their correct schema.
iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkPositionDeltaWrite.scala	Replaces hard-coded positionDeleteSparkType/Map.empty in prepareWrite with dataSparkTypeWithFieldIds/writeProps; adds dataWriterSchemaFor to correctly pick DELETE's path-pos schema vs the Iceberg data schema for other commands.
sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala	Expands Iceberg cudfWrite sig to nested STRUCT/ARRAY/MAP/BINARY; allows map-of-map in MapFromArrays; adds BINARY to ExpandExec.
integration_tests/src/main/python/iceberg/init.py	Adds materialize_parquet_source helper, assert_no_cpu_project_exec, a new iceberg_nested_write_gens_list, and parquet_write_corrected_conf; all follow existing patterns.
integration_tests/src/main/python/iceberg/iceberg_ctas_test.py	Converts fallback-only tests to GPU-correctness tests using pre-materialized Parquet sources; adds focused binary/array CTAS coverage.
integration_tests/src/main/python/iceberg/iceberg_append_test.py	Replaces fallback-only append tests with GPU-correctness tests; promotes _fallback variants to positive GPU tests with assert_no_cpu_project_exec check.
tests/src/test/spark350/scala/org/apache/iceberg/spark/GpuTypeToSparkTypeSuite.scala	New Scala unit tests covering nestedMetadataJson/appendNestedFieldIdMetadata for primitives, flat lists/maps, nested list-of-list, map-of-list, struct-of-struct, and the full toSparkType round-trip.

Sequence Diagram

sequenceDiagram
    participant IcebergSchema as Iceberg Schema
    participant GpuTypeToSparkType
    participant StructField as Spark StructField (metadata)
    participant writerOptionsFromField as SchemaUtils.writerOptionsFromField
    participant setParquetFieldId as setParquetFieldId (reflection)
    participant ColumnWriterOptions as cuDF ColumnWriterOptions
    participant ParquetFile as Parquet File

    IcebergSchema->>GpuTypeToSparkType: toSparkType(schema)
    GpuTypeToSparkType->>GpuTypeToSparkType: appendNestedFieldIdMetadata(fieldType)
    GpuTypeToSparkType->>StructField: attach {parquet.field.id, rapids.parquet.list.element.field.id, ...}
    Note over StructField: Top-level field carries nested element/key/value IDs as side-channel metadata
    StructField->>writerOptionsFromField: fieldMeta with nested IDs
    writerOptionsFromField->>writerOptionsFromField: nestedFieldMetadata() → child Metadata
    writerOptionsFromField->>ColumnWriterOptions: build() element/key/value options
    writerOptionsFromField->>setParquetFieldId: setParquetFieldId(listOptions/mapOptions, parentId)
    setParquetFieldId->>ColumnWriterOptions: reflection: hasParquetFieldId=true, parquetFieldId=N
    ColumnWriterOptions->>ParquetFile: Write with Parquet field IDs preserved

Loading

_{Reviews (8): Last reviewed commit: "Update docs" | Re-trigger Greptile}

[liurenjie1024]

Thanks @res-life for this pr. One missing point is that I hope you can do a careful check in the dml tests, since there are some fallbacks no longer necessary.

[res-life]

build

[res-life]

build

[res-life]

build