Enable optimized S3 tail reads for Iceberg Parquet footers (#15384)

### 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
- [x] No user-facing change

Testing
- [ ] Added or modified tests to cover new code paths
- [x] 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
- [x] Not required

---------

Signed-off-by: Ray Liu <liurenjie2008@gmail.com>

Author: liurenjie1024

iceberg/common/src/main/java/com/nvidia/spark/rapids/fileio/iceberg/IcebergFileIO.java

@@ -49,8 +49,17 @@ public IcebergFileIO(FileIO delegate) {
 
 
   @Override
-  public RapidsInputFile newInputFile(String path) throws IOException {
-    InputFile inputFile = delegate.newInputFile(path);
+  public IcebergInputFile newInputFile(String path) throws IOException {
+    return newInputFile(delegate.newInputFile(path));
+  }
+
+  /**
+   * Wraps an existing Iceberg input file, preserving any decryption performed by Iceberg.
+   *
+   * @param inputFile the Iceberg input file to wrap
+   */
+  public IcebergInputFile newInputFile(InputFile inputFile) {
+    Objects.requireNonNull(inputFile, "inputFile can't be null");
     return IcebergS3InputFile.maybeCreate(inputFile, delegate);
   }
 

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

@@ -23,7 +23,6 @@
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
 import com.nvidia.spark.rapids.iceberg.ShimUtils;
 import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
-import com.nvidia.spark.rapids.jni.fileio.SeekableInputStream;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.spark.TaskContext;
@@ -33,33 +32,31 @@
 
 import java.io.IOException;
 import java.util.List;
-import java.util.OptionalLong;
 
 /**
  * S3-backed {@link RapidsInputFile} that delegates byte-range reads to
  * {@link IcebergS3RangeCopier}. The supplied {@link FileIO} is only used for
  * its property map and any per-prefix storage-credential overlays.
  */
-public final class IcebergS3InputFile implements RapidsInputFile {
+public final class IcebergS3InputFile extends IcebergInputFile {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergS3InputFile.class);
 
-  private final IcebergInputFile delegate;
   private final String s3Bucket;
   private final String s3Key;
   private final IcebergS3Client icebergS3Client;
 
   private IcebergS3InputFile(
-      IcebergInputFile delegate,
+      InputFile delegate,
       String s3Bucket,
       String s3Key,
       IcebergS3Client icebergS3Client) {
-    this.delegate = delegate;
+    super(delegate);
     this.s3Bucket = s3Bucket;
     this.s3Key = s3Key;
     this.icebergS3Client = icebergS3Client;
   }
 
-  public static RapidsInputFile maybeCreate(InputFile inputFile, FileIO fileIO) {
+  public static IcebergInputFile maybeCreate(InputFile inputFile, FileIO fileIO) {
     // When the gating conf is off (or the file is not an S3 file), return the
     // default IcebergInputFile so the standard Iceberg SeekableInputStream path is used.
     IcebergInputFile delegate = new IcebergInputFile(inputFile);
@@ -86,36 +83,7 @@ public static RapidsInputFile maybeCreate(InputFile inputFile, FileIO fileIO) {
       return delegate;
     }
     LOG.debug("IcebergS3RangeCopier path active for {}", inputFile.location());
-    return new IcebergS3InputFile(delegate, s3Bucket, s3Key, icebergS3Client);
-  }
-
-  @Override
-  public String path() {
-    return delegate.path();
-  }
-
-  @Override
-  public long getLength() throws IOException {
-    return delegate.getLength();
-  }
-
-  @Override
-  public OptionalLong getLastModificationTime() throws IOException {
-    return delegate.getLastModificationTime();
-  }
-
-  @Override
-  public SeekableInputStream open() throws IOException {
-    return delegate.open();
-  }
-
-  /**
-   * Returns the underlying Iceberg {@link InputFile}, matching
-   * {@link IcebergInputFile#getDelegate()} for use by iceberg-internal
-   * code paths that need direct access to the iceberg API.
-   */
-  public InputFile getDelegate() {
-    return delegate.getDelegate();
+    return new IcebergS3InputFile(inputFile, s3Bucket, s3Key, icebergS3Client);
   }
 
   @Override
@@ -140,5 +108,11 @@ public void readTail(long length, HostMemoryBuffer output) throws IOException {
     }
     IcebergS3RangeCopier.copyTailToHMB(
         icebergS3Client, output, s3Bucket, s3Key, length, /*dstOffset*/ 0L);
+    LOG.debug(
+        "PerfIO S3 Iceberg readTail suffix-range GET completed: uri=s3://{}/{}, "
+            + "range=bytes=-{}",
+        s3Bucket,
+        s3Key,
+        length);
   }
 }

iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuIcebergPartitionReader.scala

@@ -20,7 +20,7 @@ import scala.collection.JavaConverters._
 
 import com.nvidia.spark.rapids.GpuMetric
 import com.nvidia.spark.rapids.MapUtil.toMapStrict
-import com.nvidia.spark.rapids.fileio.iceberg.{IcebergFileIO, IcebergInputFile}
+import com.nvidia.spark.rapids.fileio.iceberg.IcebergFileIO
 import com.nvidia.spark.rapids.iceberg.ShimUtils
 import com.nvidia.spark.rapids.iceberg.ShimUtils.locationOf
 import com.nvidia.spark.rapids.iceberg.data.GpuDeleteFilter
@@ -110,7 +110,7 @@ class GpuIcebergPartitionReader(private val task: GpuSparkInputPartition,
     val inputFiles = table.encryption()
       .decrypt(encryptedFiles.asJava)
       .asScala
-      .map(f => f.location() -> new IcebergInputFile(f))
+      .map(f => f.location() -> rapidsFileIO.newInputFile(f))
       .toMap
 
     val taskMap = toMapStrict(tasks.map(t => {

sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala

@@ -570,7 +570,7 @@ protected case class GpuParquetFileFilterHandler(
     if (fileIO.isInstanceOf[HadoopFileIO]) {
       // We should remove this after https://github.com/NVIDIA/spark-rapids/issues/13306 is
       // implemented.
-      val result = PerfIO.readParquetFooterBuffer(filePath, conf, verifyParquetMagic)
+      val result = PerfIO.readParquetFooterBuffer(filePath, conf, verifyParquetMagic _)
       val scheme = filePath.toUri.getScheme
       if (scheme != null && scheme.startsWith("s3")) {
         GpuTaskMetrics.get.recordPerfioS3BackendOnce()

sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/ParquetFooterUtils.scala

@@ -20,13 +20,11 @@ import java.nio.charset.StandardCharsets
 import java.util.Arrays
 
 import ai.rapids.cudf.HostMemoryBuffer
-import com.nvidia.spark.rapids.{GpuMetric, NoopMetric, NvtxRegistry, RapidsConf}
-import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
+import com.nvidia.spark.rapids.{GpuMetric, NoopMetric, NvtxRegistry, PerfIO, RapidsConf}
+import com.nvidia.spark.rapids.Arm.closeOnExcept
 import com.nvidia.spark.rapids.filecache.FileCache
 import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile
 import org.apache.hadoop.fs.Path
-import org.apache.hadoop.io.IOUtils
-import org.apache.parquet.bytes.BytesUtils.readIntLittleEndian
 import org.apache.parquet.hadoop.ParquetFileWriter.MAGIC
 
 /**
@@ -80,36 +78,8 @@ object ParquetFooterUtils {
   def readFooterBufferFromInputFile(
       inputFile: RapidsInputFile,
       filePath: Path): HostMemoryBuffer = {
-    val fileLen = inputFile.getLength
-    if (fileLen < MAGIC.length + FooterLengthSize + MAGIC.length) {
-      throw new RuntimeException(s"$filePath is not a Parquet file (too small length: $fileLen )")
-    }
-    val footerLengthIndex = fileLen - FooterLengthSize - MAGIC.length
-    withResource(inputFile.open()) { inputStream =>
-      NvtxRegistry.PARQUET_READ_FOOTER_BYTES {
-        inputStream.seek(footerLengthIndex)
-        val footerLength = readIntLittleEndian(inputStream)
-        val magic = new Array[Byte](MAGIC.length)
-        IOUtils.readFully(inputStream, magic, 0, magic.length)
-        verifyParquetMagic(filePath, magic)
-        val fIdx = footerIndex(filePath, fileLen, footerLength, FooterLengthSize)
-        val tailBytes = Math.toIntExact(fileLen - fIdx)
-        closeOnExcept(HostMemoryBuffer.allocate(tailBytes + MAGIC.length, false)) { outBuffer =>
-          outBuffer.setBytes(0, MAGIC, 0, MAGIC.length)
-          inputStream.seek(fIdx)
-          val tmp = new Array[Byte](4096)
-          var written = MAGIC.length.toLong
-          var bytesLeft = tailBytes
-          while (bytesLeft > 0) {
-            val toRead = math.min(bytesLeft, tmp.length)
-            IOUtils.readFully(inputStream, tmp, 0, toRead)
-            outBuffer.setBytes(written, tmp, 0, toRead)
-            written += toRead
-            bytesLeft -= toRead
-          }
-          outBuffer
-        }
-      }
+    NvtxRegistry.PARQUET_READ_FOOTER_BYTES {
+      PerfIO.readParquetFooterBuffer(inputFile, filePath, verifyParquetMagic)
     }
   }