Prefetch Iceberg Parquet footers with a suffix read

Signed-off-by: Zach Puller <zpuller@nvidia.com>

Author: zpuller

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java

@@ -20,6 +20,7 @@
 import com.nvidia.spark.rapids.NoopMetric$;
 import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
+import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.FileScanTask;
@@ -93,6 +94,7 @@ public interface IcebergShimUtils {
      */
     default ParquetFileReader openParquetReader(
             IcebergInputFile inputFile,
+            RapidsInputFile footerFile,
             Path filePath,
             ParquetReadOptions options,
             scala.collection.immutable.Map<String, GpuMetric> metrics) throws IOException {

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java

@@ -20,6 +20,7 @@
 import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.ShimLoader;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
+import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
 
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.ContentFile;
@@ -69,10 +70,11 @@ public static Map<String, Map<String, String>> storageCredentialOverlays(FileIO
 
     public static ParquetFileReader openParquetReader(
             IcebergInputFile inputFile,
+            RapidsInputFile footerFile,
             Path filePath,
             ParquetReadOptions options,
             scala.collection.immutable.Map<String, GpuMetric> metrics) throws IOException {
-        return IMPL.openParquetReader(inputFile, filePath, options, metrics);
+        return IMPL.openParquetReader(inputFile, footerFile, filePath, options, metrics);
     }
 
     public static GpuSparkScan newCopyOnWriteScan(

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

@@ -127,12 +127,18 @@ public void readVectored(HostMemoryBuffer output, List<CopyRange> copyRanges)
    */
   @Override
   public void readTail(long length, HostMemoryBuffer output) throws IOException {
+    readTail(length, output, 0L);
+  }
+
+  public long readTail(long length, HostMemoryBuffer output, long outputOffset)
+      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, outputOffset);
   }
 }

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/data/GpuDeleteFilter.scala

@@ -415,4 +415,3 @@ private case class DeleteFilterContext(
       joinTime)
   }
 }
-

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/data/GpuDeleteLoader.scala

@@ -117,4 +117,4 @@ class DefaultDeleteLoader(
       case SingleFile => SingleFile
     }
   }
-}
+}

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/reader.scala

@@ -25,6 +25,7 @@ import scala.collection.JavaConverters._
 import com.nvidia.spark.rapids.{CombineConf, DateTimeRebaseCorrected, GpuMetric, ThreadPoolConfBuilder}
 import com.nvidia.spark.rapids.Arm.withResource
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile
+import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile
 import com.nvidia.spark.rapids.iceberg.parquet.converter.FromIcebergShaded._
 import com.nvidia.spark.rapids.parquet.{GpuParquetUtils, ParquetFileInfoWithBlockMeta}
 import com.nvidia.spark.rapids.shims.PartitionedFileUtilsShim
@@ -52,8 +53,10 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
 case class IcebergPartitionedFile(
     file: IcebergInputFile,
     split: Option[(Long, Long)] = None,
-    filter: Option[Expression] = None) {
+    filter: Option[Expression] = None,
+    footerFile: RapidsInputFile = null) {
 
+  private lazy val footerInputFile = Option(footerFile).getOrElse(file)
   lazy val urlEncodedPath: String = new Path(file.getDelegate.location()).toUri.toString
   lazy val path: Path = new Path(new URI(urlEncodedPath))
 
@@ -63,7 +66,7 @@ case class IcebergPartitionedFile(
 
   def newReader(metrics: Map[String, GpuMetric] = Map.empty): ParquetFileReader = {
     try {
-      GpuParquetIO.openReader(file, path, parquetReadOptions, metrics)
+      GpuParquetIO.openReader(file, footerInputFile, path, parquetReadOptions, metrics)
     } catch {
       case e: IOException =>
         throw new UncheckedIOException(s"Failed to newInputFile Parquet file: " +
@@ -304,4 +307,4 @@ object GpuIcebergParquetReader {
     }
     optionsBuilder.build
   }
-}
+}

iceberg/common/src/main/scala/org/apache/iceberg/parquet/GpuParquetIO.scala

@@ -17,6 +17,7 @@
 package org.apache.iceberg.parquet
 
 import com.nvidia.spark.rapids.GpuMetric
+import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile
 import com.nvidia.spark.rapids.iceberg.ShimUtils
 import org.apache.hadoop.fs.Path
@@ -38,9 +39,10 @@ object GpuParquetIO {
    */
   def openReader(
       inputFile: IcebergInputFile,
+      footerFile: RapidsInputFile,
       filePath: Path,
       options: ParquetReadOptions,
       metrics: Map[String, GpuMetric]): ParquetFileReader = {
-    ShimUtils.openParquetReader(inputFile, filePath, options, metrics)
+    ShimUtils.openParquetReader(inputFile, footerFile, filePath, options, metrics)
   }
 }

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

@@ -26,6 +26,7 @@ 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
 
@@ -117,7 +118,8 @@ class GpuIcebergPartitionReader(private val task: GpuSparkInputPartition,
       val file = inputFiles(locationOf(t.file()))
       val icebergFile = IcebergPartitionedFile(file,
         Some((t.start(), t.length())),
-        Some(t.residual()))
+        Some(t.residual()),
+        IcebergS3InputFile.maybeCreate(file.getDelegate, fileIO))
 
       icebergFile -> t
     }))

iceberg/iceberg-1-10-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg110x/ShimUtilsImpl.java

@@ -19,6 +19,7 @@
 import com.nvidia.spark.rapids.GpuMetric;
 import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
+import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
 import com.nvidia.spark.rapids.iceberg.IcebergShimUtils;
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.*;
@@ -73,10 +74,11 @@ public Map<String, Map<String, String>> storageCredentialOverlays(FileIO fileIO)
     @Override
     public ParquetFileReader openParquetReader(
             IcebergInputFile inputFile,
+            RapidsInputFile footerFile,
             Path filePath,
             ParquetReadOptions options,
             scala.collection.immutable.Map<String, GpuMetric> metrics) throws IOException {
-        return GpuParquetIOShim.openReader(inputFile, filePath, options, metrics);
+        return GpuParquetIOShim.openReader(inputFile, footerFile, filePath, options, metrics);
     }
 
     @Override

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

@@ -17,11 +17,13 @@
 package com.nvidia.spark.rapids.iceberg.iceberg110x
 
 import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
-import com.nvidia.spark.rapids.GpuMetric
+import com.nvidia.spark.rapids.{GpuMetric, PerfIO}
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile
 import com.nvidia.spark.rapids.iceberg.parquet.converter.ToIcebergShaded
+import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile
 import com.nvidia.spark.rapids.parquet.{HMBInputFile, ParquetFooterUtils}
 import org.apache.hadoop.fs.Path
+import org.apache.iceberg.aws.s3.IcebergS3InputFile
 import org.apache.iceberg.parquet.GpuParquetIO
 import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions
 import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader
@@ -36,12 +38,22 @@ import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader
 object GpuParquetIOShim {
   def openReader(
       inputFile: IcebergInputFile,
+      footerFile: RapidsInputFile,
       filePath: Path,
       options: ParquetReadOptions,
       metrics: Map[String, GpuMetric]): ParquetFileReader = {
     val metadata = withResource(ParquetFooterUtils.getFooterBuffer(
         inputFile, metrics,
-        ParquetFooterUtils.readFooterBufferFromInputFile(inputFile, filePath))) { hmb =>
+        footerFile match {
+          case s3File: IcebergS3InputFile =>
+            PerfIO.readParquetFooterBufferFromTail(
+              filePath,
+              (length, output, outputOffset) =>
+                s3File.readTail(length, output, outputOffset),
+              ParquetFooterUtils.verifyParquetMagic)
+          case _ =>
+            ParquetFooterUtils.readFooterBufferFromInputFile(footerFile, filePath)
+        })) { hmb =>
       val shadedHmbFile = ToIcebergShaded.shade(new HMBInputFile(hmb))
       withResource(shadedHmbFile.newStream()) { hmbStream =>
         ParquetFileReader.readFooter(shadedHmbFile, options, hmbStream)