Prefetch Iceberg Parquet footers with a suffix read

Author: zpuller

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

@@ -20,6 +20,7 @@
 import com.nvidia.spark.rapids.IcebergS3RangeCopier;
 import com.nvidia.spark.rapids.IcebergS3RangeCopier.IcebergS3Client;
 import com.nvidia.spark.rapids.fileio.RapidsInputFiles;
+import com.nvidia.spark.rapids.fileio.SizedTailInputFile;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
 import com.nvidia.spark.rapids.iceberg.ShimUtils;
 import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
@@ -43,7 +44,7 @@
  *
  * <p>The package-private S3 file access is isolated in {@link IcebergS3InputFileAccess}.
  */
-public final class IcebergS3InputFile implements RapidsInputFile {
+public final class IcebergS3InputFile extends IcebergInputFile implements SizedTailInputFile {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergS3InputFile.class);
 
   private final IcebergInputFile delegate;
@@ -52,12 +53,13 @@ public final class IcebergS3InputFile implements RapidsInputFile {
 
   private IcebergS3InputFile(
       IcebergInputFile delegate, URI s3Uri, IcebergS3Client icebergS3Client) {
+    super(delegate.getDelegate());
     this.delegate = delegate;
     this.s3Uri = s3Uri;
     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);
@@ -127,12 +129,18 @@ public void readVectored(HostMemoryBuffer output, List<CopyRange> copyRanges)
    */
   @Override
   public void readTail(long length, HostMemoryBuffer output) throws IOException {
+    readTailAndGetSize(length, output);
+  }
+
+  @Override
+  public long readTailAndGetSize(long length, HostMemoryBuffer output) throws IOException {
     if (length == 0) {
-      return;
+      return 0;
     }
     if (length < 0) {
       throw new IllegalArgumentException("length must be non-negative");
     }
-    IcebergS3RangeCopier.copyTailToHMB(icebergS3Client, output, s3Uri, length, /*dstOffset*/ 0L);
+    return IcebergS3RangeCopier.copyTailToHMB(
+        icebergS3Client, output, s3Uri, length, /*dstOffset*/ 0L);
   }
 }

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

@@ -20,12 +20,13 @@ 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
 import com.nvidia.spark.rapids.iceberg.parquet._
 import org.apache.iceberg._
+import org.apache.iceberg.aws.s3.IcebergS3InputFile
 import org.apache.iceberg.encryption.EncryptedFiles
 import org.apache.iceberg.mapping.NameMappingParser
 
@@ -110,7 +111,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() -> IcebergS3InputFile.maybeCreate(f, fileIO))
       .toMap
 
     val taskMap = toMapStrict(tasks.map(t => {

iceberg/iceberg-1-10-x/src/main/scala/com/nvidia/spark/rapids/iceberg/iceberg110x/GpuParquetIOShim.scala

@@ -41,7 +41,8 @@ object GpuParquetIOShim {
       metrics: Map[String, GpuMetric]): ParquetFileReader = {
     val metadata = withResource(ParquetFooterUtils.getFooterBuffer(
         inputFile, metrics,
-        ParquetFooterUtils.readFooterBufferFromInputFile(inputFile, filePath))) { hmb =>
+        ParquetFooterUtils.readFooterBufferFromInputFileWithSuffixPrefetch(inputFile, filePath))) {
+      hmb =>
       val shadedHmbFile = ToIcebergShaded.shade(new HMBInputFile(hmb))
       withResource(shadedHmbFile.newStream()) { hmbStream =>
         ParquetFileReader.readFooter(shadedHmbFile, options, hmbStream)

iceberg/iceberg-1-11-x/src/main/scala/com/nvidia/spark/rapids/iceberg/iceberg111x/GpuParquetIOShim.scala

@@ -41,7 +41,8 @@ object GpuParquetIOShim {
       metrics: Map[String, GpuMetric]): ParquetFileReader = {
     val metadata = withResource(ParquetFooterUtils.getFooterBuffer(
         inputFile, metrics,
-        ParquetFooterUtils.readFooterBufferFromInputFile(inputFile, filePath))) { hmb =>
+        ParquetFooterUtils.readFooterBufferFromInputFileWithSuffixPrefetch(inputFile, filePath))) {
+      hmb =>
       val shadedHmbFile = ToIcebergShaded.shade(new HMBInputFile(hmb))
       withResource(shadedHmbFile.newStream()) { hmbStream =>
         ParquetFileReader.readFooter(shadedHmbFile, options, hmbStream)

sql-plugin/src/main/java/com/nvidia/spark/rapids/fileio/SizedTailInputFile.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nvidia.spark.rapids.fileio;
+
+import ai.rapids.cudf.HostMemoryBuffer;
+
+import java.io.IOException;
+
+/**
+ * Optional capability for input files whose tail read can return fewer bytes than requested.
+ * The bytes are written starting at offset zero in {@code output}.
+ */
+public interface SizedTailInputFile {
+  long readTailAndGetSize(long length, HostMemoryBuffer output) throws IOException;
+}

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

@@ -23,6 +23,7 @@ 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.filecache.FileCache
+import com.nvidia.spark.rapids.fileio.SizedTailInputFile
 import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.io.IOUtils
@@ -38,6 +39,9 @@ import org.apache.parquet.hadoop.ParquetFileWriter.MAGIC
  */
 object ParquetFooterUtils {
   val FooterLengthSize: Int = java.lang.Integer.BYTES
+  private val IcebergSuffixPrefetchSizeKey =
+    "spark.rapids.iceberg.parquet.footer.suffixPrefetchSize"
+  private val DefaultIcebergSuffixPrefetchSize = 0L
   private val ParquetMagicEncrypted = "PARE".getBytes(StandardCharsets.US_ASCII)
 
   def verifyParquetMagic(filePath: Path, magic: Array[Byte]): Unit = {
@@ -113,6 +117,53 @@ object ParquetFooterUtils {
     }
   }
 
+  /**
+   * Read one configurable suffix and parse the footer from it. For Iceberg S3 PerfIO this
+   * replaces the footer-length request and following footer request with a single suffix GET.
+   * Unsupported inputs and footers larger than the prefetched suffix use the existing path.
+   */
+  def readFooterBufferFromInputFileWithSuffixPrefetch(
+      inputFile: RapidsInputFile,
+      filePath: Path): HostMemoryBuffer = {
+    val configuredSize = Option(org.apache.spark.SparkEnv.get)
+      .map(_.conf.getSizeAsBytes(
+        IcebergSuffixPrefetchSizeKey, DefaultIcebergSuffixPrefetchSize.toString))
+      .getOrElse(DefaultIcebergSuffixPrefetchSize)
+    if (configuredSize <= 0 || !inputFile.isInstanceOf[SizedTailInputFile]) {
+      return readFooterBufferFromInputFile(inputFile, filePath)
+    }
+
+    val prefetchSize = Math.toIntExact(
+      math.max(configuredSize, FooterLengthSize + MAGIC.length))
+    try {
+      withResource(HostMemoryBuffer.allocate(prefetchSize, false)) { suffix =>
+        val actualSize = Math.toIntExact(
+          inputFile.asInstanceOf[SizedTailInputFile].readTailAndGetSize(prefetchSize, suffix))
+        if (actualSize < FooterLengthSize + MAGIC.length) {
+          return readFooterBufferFromInputFile(inputFile, filePath)
+        }
+
+        val footerLengthOffset = actualSize - FooterLengthSize - MAGIC.length
+        val magic = readBytesFromBuffer(suffix, actualSize - MAGIC.length, MAGIC.length)
+        verifyParquetMagic(filePath, magic)
+        val footerLengthBytes = readBytesFromBuffer(suffix, footerLengthOffset, FooterLengthSize)
+        val footerLength = readIntLittleEndian(footerLengthBytes, 0)
+        val tailLength = footerLength + FooterLengthSize + MAGIC.length
+        if (footerLength <= 0 || tailLength > actualSize) {
+          return readFooterBufferFromInputFile(inputFile, filePath)
+        }
+
+        closeOnExcept(HostMemoryBuffer.allocate(tailLength + MAGIC.length, false)) { out =>
+          out.setBytes(0, MAGIC, 0, MAGIC.length)
+          out.copyFromHostBuffer(MAGIC.length, suffix, actualSize - tailLength, tailLength)
+          out
+        }
+      }
+    } catch {
+      case _: java.io.IOException => readFooterBufferFromInputFile(inputFile, filePath)
+    }
+  }
+
   /**
    * Return a framed footer buffer (`MAGIC + footer + footerLen + MAGIC`) for `inputFile`,
    * serving from `FileCache` when present and populating the cache on miss. The `readFooterBuffer`