Enable optimized S3 tail reads for Iceberg Parquet footers

[liurenjie1024]

Description

Fixes #15363

Parquet footer reads through Iceberg currently open and seek an input stream instead of using RapidsInputFile.readTail. This bypasses the optimized Iceberg S3 suffix-range path and adds an avoidable length/seek round trip.

This change:

• Reads the Parquet trailer with RapidsInputFile.readTail and the footer body with readVectored.
• Makes IcebergS3InputFile an IcebergInputFile subclass backed by the original org.apache.iceberg.io.InputFile, overriding only the S3-specific vectored and tail reads.
• Routes decrypted Iceberg input files through IcebergFileIO, preserving Iceberg decryption while enabling optimized S3 reads.
• Emits a debug message after a suffix-range tail read completes.

Testing

• Spark 4.0.2 compilation passed for the Iceberg 1.10 and 1.11 modules: mvn -Dbuildver=402 -DskipTests -pl iceberg/iceberg-1-10-x,iceberg/iceberg-1-11-x -am compile
• Existing Parquet and Iceberg scan coverage includes ParquetScanSuite, ParquetFormatScanSuite, and integration_tests/src/main/python/iceberg/iceberg_test.py.
• An EMR GPU table scan completed successfully, and executor logs confirmed that Iceberg readTail used an S3 suffix-range GET.

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
• 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

Routes Iceberg Parquet footer reads through optimized RAPIDS tail and vectored I/O while preserving decrypted Iceberg input files.

• Adds an InputFile-wrapping overload to IcebergFileIO.
• Makes the S3-specific input file inherit the standard Iceberg wrapper and override only optimized range operations.
• Wraps decrypted data and delete files through the optimized file-I/O path.
• Reuses PerfIO for shared Parquet footer reading and adds completion-level debug logging.

Confidence Score: 5/5

The PR appears safe to merge.

No blocking failure remains.

Important Files Changed

Filename	Overview
iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergFileIO.java	Adds wrapping for existing Iceberg input files so decrypted delegates can retain Iceberg behavior while selecting optimized S3 access.
iceberg/common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFile.java	Refactors the optimized S3 input into an IcebergInputFile subclass and adds suffix-tail completion logging.
iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuIcebergPartitionReader.scala	Routes decrypted data and delete input files through IcebergFileIO wrapping.
sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala	Reads Parquet trailers through readTail and footer bodies through readVectored.
sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/ParquetFooterUtils.scala	Replaces the local stream-and-seek footer implementation with the shared PerfIO footer reader.

Sequence Diagram

sequenceDiagram
  participant Reader as GPU Iceberg Reader
  participant FileIO as IcebergFileIO
  participant Input as IcebergS3InputFile
  participant S3 as Amazon S3
  Reader->>FileIO: Wrap decrypted InputFile
  FileIO->>Input: maybeCreate(inputFile, fileIO)
  Reader->>Input: readTail(trailerLength)
  Input->>S3: "GET Range bytes=-N"
  S3-->>Input: Parquet trailer
  Reader->>Input: readVectored(footer range)
  Input->>S3: GET footer byte range
  S3-->>Reader: Framed Parquet footer

Loading

_{Reviews (10): Last reviewed commit: "Fix build break" | Re-trigger Greptile}

[nvauto]

NOTE: release/26.08 has been created from main. Please retarget your PR to release/26.08 if it should be included in the release.

[liurenjie1024]

blocked by #15360

[res-life]

SHOULD FIX — Remove the redundant delegate field

iceberg/common/src/main/java/org/apache/iceberg/aws/s3/IcebergS3InputFile.java:44

IcebergS3InputFile stores delegate in a private field that is never read. The constructor already passes the same InputFile to IcebergInputFile at line 54, and the superclass stores it for the inherited operations. Remove only the redundant field declaration and constructor assignment; keep the constructor parameter and super(delegate) call.

@@ -41,7 +41,6 @@ public final class IcebergS3InputFile extends IcebergInputFile {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergS3InputFile.class);
 
-  private final InputFile delegate;
   private final String s3Bucket;
   private final String s3Key;
   private final IcebergS3Client icebergS3Client;
@@ -52,7 +51,6 @@ public final class IcebergS3InputFile extends IcebergInputFile {
       String s3Key,
       IcebergS3Client icebergS3Client) {
     super(delegate);
-    this.delegate = delegate;
     this.s3Bucket = s3Bucket;
     this.s3Key = s3Key;
     this.icebergS3Client = icebergS3Client;

[liurenjie1024]

build

[res-life]

LGTM

[res-life]

build

[sameerz]

build

[liurenjie1024]

build