diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index f867d4bc9b..c90d697bf7 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -19,6 +19,9 @@
 
 package org.apache.comet.iceberg
 
+import java.lang.reflect.Method
+import java.util.concurrent.ConcurrentHashMap
+
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.SparkSession
 
@@ -139,24 +142,102 @@ object IcebergReflection extends Logging {
    */
   def loadClass(className: String): Class[_] = ClassLoaders.loadClass(className)
 
+  /**
+   * Methods resolved by [[findMethod]], [[getDeclaredMethod]] and [[findMethodInHierarchy]],
+   * keyed by the class the lookup started from and then by the lookup itself.
+   *
+   * `Class.getMethod` linearly scans the class's public methods and returns a fresh defensive
+   * copy of the `Method` on every call. Comet resolves the same handful of Iceberg accessors once
+   * per file scan task, and again per partition field and per delete file, so planning a scan
+   * over a table with many files does O(files) reflective lookups that all resolve to the same
+   * few methods, repeated for every AQE stage.
+   *
+   * Misses are cached too, which matters most for [[extractFileLocation]]: it probes for
+   * `location()` on every file, and Iceberg versions that only have `path()` would otherwise
+   * construct a `NoSuchMethodException`, stack trace and all, per file.
+   *
+   * A `ClassValue` keys the cache on the class object itself, so entries are reclaimed with the
+   * class and a cached method never pins a classloader that Spark has discarded.
+   */
+  private val methodCache: ClassValue[ConcurrentHashMap[String, Option[Method]]] =
+    new ClassValue[ConcurrentHashMap[String, Option[Method]]] {
+      override protected def computeValue(
+          clazz: Class[_]): ConcurrentHashMap[String, Option[Method]] =
+        new ConcurrentHashMap[String, Option[Method]]()
+    }
+
+  private def cachedLookup(clazz: Class[_], key: String)(
+      resolve: => Option[Method]): Option[Method] = {
+    val perClass = methodCache.get(clazz)
+    // Read first: computeIfAbsent allocates the mapping function and can lock the bin even for a
+    // hit, and these lookups are almost always hits.
+    val cached = perClass.get(key)
+    if (cached != null) cached else perClass.computeIfAbsent(key, _ => resolve)
+  }
+
+  private def lookupKey(methodName: String, paramTypes: Seq[Class[_]]): String =
+    if (paramTypes.isEmpty) methodName
+    else paramTypes.map(_.getName).mkString(methodName + "(", ",", ")")
+
+  private def missing(clazz: Class[_], methodName: String, paramTypes: Seq[Class[_]]): Nothing =
+    throw new NoSuchMethodException(s"${clazz.getName}.${lookupKey(methodName, paramTypes)}")
+
+  /**
+   * Suppresses access checks so the method can be invoked when its declaring class is
+   * package-private, as Iceberg's concrete file/task/term implementations are. Runs once, when
+   * the method is first resolved. A JVM that refuses (a class in a module that is exported but
+   * not open) leaves the method usable for the public-class case, so the refusal is not fatal
+   * here.
+   */
+  private def makeAccessible(method: Method): Method = {
+    try method.setAccessible(true)
+    catch { case _: RuntimeException => }
+    method
+  }
+
+  private def declaredMethod(clazz: Class[_], methodName: String): Option[Method] =
+    try Some(makeAccessible(clazz.getDeclaredMethod(methodName)))
+    catch { case _: NoSuchMethodException => None }
+
+  /**
+   * Cached `Class.getMethod`, returning None instead of throwing when the method is absent. The
+   * resolved method has access checks suppressed (see [[makeAccessible]]).
+   */
+  def findMethod(clazz: Class[_], methodName: String, paramTypes: Class[_]*): Option[Method] =
+    cachedLookup(clazz, lookupKey(methodName, paramTypes)) {
+      try Some(makeAccessible(clazz.getMethod(methodName, paramTypes: _*)))
+      catch { case _: NoSuchMethodException => None }
+    }
+
+  /**
+   * Cached `Class.getMethod`, throwing `NoSuchMethodException` when the method is absent, like
+   * the JDK call it replaces.
+   */
+  def getMethod(clazz: Class[_], methodName: String, paramTypes: Class[_]*): Method =
+    findMethod(clazz, methodName, paramTypes: _*).getOrElse(
+      missing(clazz, methodName, paramTypes))
+
+  /**
+   * Cached `Class.getDeclaredMethod` with access checks suppressed, throwing
+   * `NoSuchMethodException` when the method is absent, like the JDK call it replaces.
+   */
+  def getDeclaredMethod(clazz: Class[_], methodName: String): Method =
+    cachedLookup(clazz, "declared:" + methodName)(declaredMethod(clazz, methodName))
+      .getOrElse(missing(clazz, methodName, Nil))
+
   /**
    * Searches through class hierarchy to find a method (including protected methods).
    */
-  def findMethodInHierarchy(
-      clazz: Class[_],
-      methodName: String): Option[java.lang.reflect.Method] = {
-    var current: Class[_] = clazz
-    while (current != null) {
-      try {
-        val method = current.getDeclaredMethod(methodName)
-        method.setAccessible(true)
-        return Some(method)
-      } catch {
-        case _: NoSuchMethodException => current = current.getSuperclass
+  def findMethodInHierarchy(clazz: Class[_], methodName: String): Option[Method] =
+    cachedLookup(clazz, "hierarchy:" + methodName) {
+      var current: Class[_] = clazz
+      var found: Option[Method] = None
+      while (found.isEmpty && current != null) {
+        found = declaredMethod(current, methodName)
+        if (found.isEmpty) current = current.getSuperclass
       }
+      found
     }
-    None
-  }
 
   /**
    * True if `clazz` or any of its superclasses has a name in `names`. Walks the already-loaded
@@ -184,16 +265,13 @@ object IcebergReflection extends Logging {
    */
   def extractFileLocation(contentFileClass: Class[_], file: Any): Option[String] = {
     try {
-      val locationMethod = contentFileClass.getMethod("location")
-      Some(locationMethod.invoke(file).asInstanceOf[String])
+      findMethod(contentFileClass, "location") match {
+        case Some(locationMethod) => Some(locationMethod.invoke(file).asInstanceOf[String])
+        case None =>
+          findMethod(contentFileClass, "path")
+            .map(_.invoke(file).asInstanceOf[CharSequence].toString)
+      }
     } catch {
-      case _: NoSuchMethodException =>
-        try {
-          val pathMethod = contentFileClass.getMethod("path")
-          Some(pathMethod.invoke(file).asInstanceOf[CharSequence].toString)
-        } catch {
-          case _: Exception => None
-        }
       case _: Exception => None
     }
   }
@@ -210,16 +288,16 @@ object IcebergReflection extends Logging {
     }
   }
 
-  /** The file format of a ContentFile (data or delete file), e.g. "PARQUET", "AVRO", "ORC". */
-  def getFileFormat(file: Any): Option[String] = {
+  /**
+   * The file format of a ContentFile (data or delete file), e.g. "PARQUET", "AVRO", "ORC".
+   *
+   * `contentFileClass` is the public ContentFile interface, which callers already hold: Iceberg's
+   * concrete file impls are package-private, so `format()` resolved on the concrete class throws
+   * IllegalAccessException when invoked.
+   */
+  def getFileFormat(contentFileClass: Class[_], file: Any): Option[String] = {
     try {
-      // Resolve format() on the public ContentFile interface. Iceberg's concrete file impls are
-      // package-private, so a method resolved on the concrete class throws IllegalAccessException
-      // when invoked.
-      // TODO callers in a loop (e.g. validateIcebergFileScanTasks) already hold a cached
-      // contentFileClass; add an overload that takes it to avoid reloading per file.
-      val contentFileClass = loadClass(ClassNames.CONTENT_FILE)
-      Some(contentFileClass.getMethod("format").invoke(file).toString)
+      findMethod(contentFileClass, "format").map(_.invoke(file).toString)
     } catch {
       case _: Exception => None
     }
@@ -279,7 +357,7 @@ object IcebergReflection extends Logging {
           } else {
             // All task groups in a stage share the same concrete class, so the per-group
             // `tasks()` lookup can be cached once instead of done N times.
-            val groupTasksMethod = groups.get(0).getClass.getMethod("tasks")
+            val groupTasksMethod = getMethod(groups.get(0).getClass, "tasks")
             val flat = new java.util.ArrayList[AnyRef]()
             groups.forEach { group =>
               val groupTasks =
@@ -335,19 +413,17 @@ object IcebergReflection extends Logging {
    */
   def getFormatVersion(table: Any): Option[Int] = {
     try {
-      val formatVersionMethod = table.getClass.getMethod("formatVersion")
+      val formatVersionMethod = getMethod(table.getClass, "formatVersion")
       Some(formatVersionMethod.invoke(table).asInstanceOf[Int])
     } catch {
       case _: NoSuchMethodException =>
         try {
           // If not directly available, access via operations/metadata
-          val opsMethod = table.getClass.getDeclaredMethod("operations")
-          opsMethod.setAccessible(true)
-          val ops = opsMethod.invoke(table)
+          val ops = getDeclaredMethod(table.getClass, "operations").invoke(table)
           findMethodInHierarchy(ops.getClass, "current")
             .flatMap { currentMethod =>
               val metadata = currentMethod.invoke(ops)
-              val formatVersionMethod = metadata.getClass.getMethod("formatVersion")
+              val formatVersionMethod = getMethod(metadata.getClass, "formatVersion")
               Some(formatVersionMethod.invoke(metadata).asInstanceOf[Int])
             }
             .orElse {
@@ -372,7 +448,7 @@ object IcebergReflection extends Logging {
    */
   def getFileIO(table: Any): Option[Any] = {
     try {
-      val ioMethod = table.getClass.getMethod("io")
+      val ioMethod = getMethod(table.getClass, "io")
       Some(ioMethod.invoke(table))
     } catch {
       case e: Exception =>
@@ -412,7 +488,7 @@ object IcebergReflection extends Logging {
    */
   def getSchema(table: Any): Option[Any] = {
     try {
-      val schemaMethod = table.getClass.getMethod("schema")
+      val schemaMethod = getMethod(table.getClass, "schema")
       Some(schemaMethod.invoke(table))
     } catch {
       case e: Exception =>
@@ -429,8 +505,7 @@ object IcebergReflection extends Logging {
   def getAllSchemas(table: Any): Seq[Any] = {
     import scala.jdk.CollectionConverters._
     try {
-      table.getClass
-        .getMethod("schemas")
+      getMethod(table.getClass, "schemas")
         .invoke(table)
         .asInstanceOf[java.util.Map[_, _]]
         .values()
@@ -446,7 +521,7 @@ object IcebergReflection extends Logging {
   /** Returns the `Types.NestedField` for `fieldId` in `schema`, or None. */
   def findFieldObject(schema: Any, fieldId: Int): Option[Any] = {
     try {
-      val findFieldMethod = schema.getClass.getMethod("findField", classOf[Int])
+      val findFieldMethod = getMethod(schema.getClass, "findField", classOf[Int])
       Option(findFieldMethod.invoke(schema, fieldId.asInstanceOf[AnyRef]))
     } catch {
       case _: Exception => None
@@ -483,8 +558,7 @@ object IcebergReflection extends Logging {
             s"Cannot resolve equality-delete field id $id in table schema history"))
       }
       val existing =
-        baseSchema.getClass
-          .getMethod("columns")
+        getMethod(baseSchema.getClass, "columns")
           .invoke(baseSchema)
           .asInstanceOf[java.util.List[_]]
       val newColumns = new java.util.ArrayList[Any](existing)
@@ -501,7 +575,7 @@ object IcebergReflection extends Logging {
    */
   def getPartitionSpec(table: Any): Option[Any] = {
     try {
-      val specMethod = table.getClass.getMethod("spec")
+      val specMethod = getMethod(table.getClass, "spec")
       Some(specMethod.invoke(table))
     } catch {
       case e: Exception =>
@@ -531,8 +605,7 @@ object IcebergReflection extends Logging {
     try {
       val tableClass = loadClass(ClassNames.TABLE)
       val partitioningClass = loadClass(ClassNames.PARTITIONING)
-      partitioningClass
-        .getMethod("partitionType", tableClass)
+      getMethod(partitioningClass, "partitionType", tableClass)
         .invoke(null, table.asInstanceOf[AnyRef])
       None
     } catch {
@@ -555,9 +628,7 @@ object IcebergReflection extends Logging {
    */
   def getTableMetadata(table: Any): Option[Any] = {
     try {
-      val operationsMethod = table.getClass.getDeclaredMethod("operations")
-      operationsMethod.setAccessible(true)
-      val operations = operationsMethod.invoke(table)
+      val operations = getDeclaredMethod(table.getClass, "operations").invoke(table)
 
       findMethodInHierarchy(operations.getClass, "current").map(_.invoke(operations)).orElse {
         logError(
@@ -589,7 +660,7 @@ object IcebergReflection extends Logging {
   def getMetadataLocation(table: Any): Option[String] = {
     getTableMetadata(table).flatMap { metadata =>
       try {
-        val metadataFileLocationMethod = metadata.getClass.getMethod("metadataFileLocation")
+        val metadataFileLocationMethod = getMethod(metadata.getClass, "metadataFileLocation")
         Some(metadataFileLocationMethod.invoke(metadata).asInstanceOf[String])
       } catch {
         case e: Exception =>
@@ -611,7 +682,7 @@ object IcebergReflection extends Logging {
   def getTableProperties(table: Any): Option[java.util.Map[String, String]] = {
     getTableMetadata(table).flatMap { metadata =>
       try {
-        val propertiesMethod = metadata.getClass.getMethod("properties")
+        val propertiesMethod = getMethod(metadata.getClass, "properties")
         Some(propertiesMethod.invoke(metadata).asInstanceOf[java.util.Map[String, String]])
       } catch {
         case e: Exception =>
@@ -634,7 +705,7 @@ object IcebergReflection extends Logging {
    *   if reflection fails (callers must handle appropriately based on context)
    */
   def getDeleteFilesFromTask(task: Any, fileScanTaskClass: Class[_]): java.util.List[_] = {
-    val deletesMethod = fileScanTaskClass.getMethod("deletes")
+    val deletesMethod = getMethod(fileScanTaskClass, "deletes")
     val deletes = deletesMethod.invoke(task).asInstanceOf[java.util.List[_]]
     if (deletes == null) new java.util.ArrayList[Any]() else deletes
   }
@@ -642,16 +713,19 @@ object IcebergReflection extends Logging {
   /**
    * Gets equality field IDs from a delete file.
    *
+   * @param deleteFileClass
+   *   The DeleteFile interface, which callers in a loop already hold
    * @param deleteFile
    *   An Iceberg DeleteFile object
    * @return
    *   List of field IDs used in equality deletes, or empty list for position deletes
    */
-  def getEqualityFieldIds(deleteFile: Any): java.util.List[_] = {
+  def getEqualityFieldIds(deleteFileClass: Class[_], deleteFile: Any): java.util.List[_] = {
     try {
-      val deleteFileClass = loadClass(ClassNames.DELETE_FILE)
-      val equalityFieldIdsMethod = deleteFileClass.getMethod("equalityFieldIds")
-      val ids = equalityFieldIdsMethod.invoke(deleteFile).asInstanceOf[java.util.List[_]]
+      val ids =
+        getMethod(deleteFileClass, "equalityFieldIds")
+          .invoke(deleteFile)
+          .asInstanceOf[java.util.List[_]]
       if (ids == null) new java.util.ArrayList[Any]() else ids
     } catch {
       case _: Exception =>
@@ -672,11 +746,11 @@ object IcebergReflection extends Logging {
    */
   def getFieldInfo(schema: Any, fieldId: Int): Option[(String, String)] = {
     try {
-      val findFieldMethod = schema.getClass.getMethod("findField", classOf[Int])
+      val findFieldMethod = getMethod(schema.getClass, "findField", classOf[Int])
       val field = findFieldMethod.invoke(schema, fieldId.asInstanceOf[AnyRef])
       if (field != null) {
-        val nameMethod = field.getClass.getMethod("name")
-        val typeMethod = field.getClass.getMethod("type")
+        val nameMethod = getMethod(field.getClass, "name")
+        val typeMethod = getMethod(field.getClass, "type")
         val fieldName = nameMethod.invoke(field).toString
         val fieldType = typeMethod.invoke(field).toString
         Some((fieldName, fieldType))
@@ -734,15 +808,15 @@ object IcebergReflection extends Logging {
   def buildFieldIdMapping(schema: Any): Map[String, Int] = {
     import scala.jdk.CollectionConverters._
     try {
-      val columnsMethod = schema.getClass.getMethod("columns")
+      val columnsMethod = getMethod(schema.getClass, "columns")
       val columns = columnsMethod.invoke(schema).asInstanceOf[java.util.List[_]]
 
       columns.asScala.flatMap { column =>
         try {
-          val nameMethod = column.getClass.getMethod("name")
+          val nameMethod = getMethod(column.getClass, "name")
           val name = nameMethod.invoke(column).asInstanceOf[String]
 
-          val fieldIdMethod = column.getClass.getMethod("fieldId")
+          val fieldIdMethod = getMethod(column.getClass, "fieldId")
           val fieldId = fieldIdMethod.invoke(column).asInstanceOf[Int]
 
           Some(name -> fieldId)
@@ -773,13 +847,12 @@ object IcebergReflection extends Logging {
   def pageIndexUnsupportedColumns(schema: Any): Set[String] = {
     import scala.jdk.CollectionConverters._
     try {
-      val columns = schema.getClass
-        .getMethod("columns")
+      val columns = getMethod(schema.getClass, "columns")
         .invoke(schema)
         .asInstanceOf[java.util.List[_]]
       columns.asScala.flatMap { column =>
-        val name = column.getClass.getMethod("name").invoke(column).asInstanceOf[String]
-        val typeStr = column.getClass.getMethod("type").invoke(column).toString
+        val name = getMethod(column.getClass, "name").invoke(column).asInstanceOf[String]
+        val typeStr = getMethod(column.getClass, "type").invoke(column).toString
         if (typeStr.startsWith("decimal(") || typeStr == "uuid" || typeStr.startsWith("fixed[") ||
           typeStr == "binary") {
           Some(name)
@@ -812,12 +885,12 @@ object IcebergReflection extends Logging {
   def validatePartitionTypes(partitionSpec: Any, schema: Any): List[(String, String, String)] = {
     import scala.jdk.CollectionConverters._
 
-    val fieldsMethod = partitionSpec.getClass.getMethod("fields")
+    val fieldsMethod = getMethod(partitionSpec.getClass, "fields")
     val fields = fieldsMethod.invoke(partitionSpec).asInstanceOf[java.util.List[_]]
 
     val partitionFieldClass = loadClass(ClassNames.PARTITION_FIELD)
-    val sourceIdMethod = partitionFieldClass.getMethod("sourceId")
-    val findFieldMethod = schema.getClass.getMethod("findField", classOf[Int])
+    val sourceIdMethod = getMethod(partitionFieldClass, "sourceId")
+    val findFieldMethod = getMethod(schema.getClass, "findField", classOf[Int])
 
     val unsupportedTypes = scala.collection.mutable.ListBuffer[(String, String, String)]()
 
@@ -826,10 +899,10 @@ object IcebergReflection extends Logging {
       val column = findFieldMethod.invoke(schema, sourceId.asInstanceOf[Object])
 
       if (column != null) {
-        val nameMethod = column.getClass.getMethod("name")
+        val nameMethod = getMethod(column.getClass, "name")
         val fieldName = nameMethod.invoke(column).asInstanceOf[String]
 
-        val typeMethod = column.getClass.getMethod("type")
+        val typeMethod = getMethod(column.getClass, "type")
         val icebergType = typeMethod.invoke(column)
         val typeStr = icebergType.toString
 
@@ -866,22 +939,21 @@ object IcebergReflection extends Logging {
   def columnsWithInitialDefault(schema: Any): List[String] = {
     import scala.jdk.CollectionConverters._
     val columns =
-      schema.getClass.getMethod("columns").invoke(schema).asInstanceOf[java.util.List[_]]
+      getMethod(schema.getClass, "columns").invoke(schema).asInstanceOf[java.util.List[_]]
     columns.asScala.flatMap(walkFieldForDefault).toList
   }
 
   private def walkFieldForDefault(field: Any): List[String] = {
     import scala.jdk.CollectionConverters._
-    val name = field.getClass.getMethod("name").invoke(field).asInstanceOf[String]
+    val name = getMethod(field.getClass, "name").invoke(field).asInstanceOf[String]
     val here =
-      if (field.getClass.getMethod("initialDefault").invoke(field) != null) List(name) else Nil
-    val fieldType = field.getClass.getMethod("type").invoke(field)
+      if (getMethod(field.getClass, "initialDefault").invoke(field) != null) List(name) else Nil
+    val fieldType = getMethod(field.getClass, "type").invoke(field)
     val nested =
-      if (fieldType.getClass.getMethod("isNestedType").invoke(fieldType).asInstanceOf[Boolean]) {
-        val nestedType = fieldType.getClass.getMethod("asNestedType").invoke(fieldType)
+      if (getMethod(fieldType.getClass, "isNestedType").invoke(fieldType).asInstanceOf[Boolean]) {
+        val nestedType = getMethod(fieldType.getClass, "asNestedType").invoke(fieldType)
         val fields =
-          nestedType.getClass
-            .getMethod("fields")
+          getMethod(nestedType.getClass, "fields")
             .invoke(nestedType)
             .asInstanceOf[java.util.List[_]]
         fields.asScala.flatMap(walkFieldForDefault).toList
@@ -901,7 +973,7 @@ object IcebergReflection extends Logging {
   def toSparkSchema(schema: Any): org.apache.spark.sql.types.StructType = {
     val sparkSchemaUtil = loadClass(ClassNames.SPARK_SCHEMA_UTIL)
     val schemaClass = loadClass(ClassNames.SCHEMA)
-    val convert = sparkSchemaUtil.getMethod("convert", schemaClass)
+    val convert = getMethod(sparkSchemaUtil, "convert", schemaClass)
     convert
       .invoke(null, schema.asInstanceOf[AnyRef])
       .asInstanceOf[org.apache.spark.sql.types.StructType]
@@ -1048,7 +1120,7 @@ object CometIcebergNativeScanMetadata extends Logging {
 
   private def invokeTableName(table: Any): Option[String] = {
     try {
-      table.getClass.getMethod("name").invoke(table) match {
+      IcebergReflection.getMethod(table.getClass, "name").invoke(table) match {
         case s: String => Some(s)
         case other if other != null => Some(other.toString)
         case null => None
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index ca917335da..a57475d2f5 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -419,7 +419,7 @@ case class CometScanRule(session: SparkSession)
                 tableOpt
                   .flatMap { table =>
                     try {
-                      val locationMethod = table.getClass.getMethod("location")
+                      val locationMethod = IcebergReflection.getMethod(table.getClass, "location")
                       val tableLocation = locationMethod.invoke(table).asInstanceOf[String]
                       Some(tableLocation)
                     } catch {
@@ -719,10 +719,14 @@ case class CometScanRule(session: SparkSession)
           try {
             if (!taskValidation.deleteFiles.isEmpty) {
               val historicSchemas = IcebergReflection.getAllSchemas(metadata.table)
+              val contentFileClass =
+                IcebergReflection.loadClass(IcebergReflection.ClassNames.CONTENT_FILE)
+              val deleteFileClass =
+                IcebergReflection.loadClass(IcebergReflection.ClassNames.DELETE_FILE)
               taskValidation.deleteFiles.asScala.foreach { deleteFile =>
                 // iceberg-rust only reads Parquet delete files. Avro/ORC positional or
                 // equality deletes must be applied by Spark.
-                IcebergReflection.getFileFormat(deleteFile) match {
+                IcebergReflection.getFileFormat(contentFileClass, deleteFile) match {
                   case Some(fmt) if fmt.equalsIgnoreCase(IcebergReflection.FileFormats.PARQUET) =>
                   case Some(fmt) =>
                     hasUnsupportedDeletes = true
@@ -736,7 +740,8 @@ case class CometScanRule(session: SparkSession)
                     fallbackReasons += "Could not determine Iceberg delete file format"
                 }
 
-                val equalityFieldIds = IcebergReflection.getEqualityFieldIds(deleteFile)
+                val equalityFieldIds =
+                  IcebergReflection.getEqualityFieldIds(deleteFileClass, deleteFile)
 
                 if (!equalityFieldIds.isEmpty) {
                   equalityFieldIds.asScala.foreach { fieldId =>
@@ -983,13 +988,13 @@ object CometScanRule extends Logging {
     val unboundPredicateClass =
       IcebergReflection.loadClass(IcebergReflection.ClassNames.UNBOUND_PREDICATE)
 
-    // Cache all method lookups outside the loop
-    val fileMethod = contentScanTaskClass.getMethod("file")
-    val formatMethod = contentFileClass.getMethod("format")
-    val pathMethod = contentFileClass.getMethod("path")
-    val residualMethod = contentScanTaskClass.getMethod("residual")
-    val deletesMethod = fileScanTaskClass.getMethod("deletes")
-    val termMethod = unboundPredicateClass.getMethod("term")
+    // Resolve all method lookups outside the loop
+    val fileMethod = IcebergReflection.getMethod(contentScanTaskClass, "file")
+    val formatMethod = IcebergReflection.getMethod(contentFileClass, "format")
+    val pathMethod = IcebergReflection.getMethod(contentFileClass, "path")
+    val residualMethod = IcebergReflection.getMethod(contentScanTaskClass, "residual")
+    val deletesMethod = IcebergReflection.getMethod(fileScanTaskClass, "deletes")
+    val termMethod = IcebergReflection.getMethod(unboundPredicateClass, "term")
 
     val supportedSchemes =
       Set("file", "s3", "s3a", "gs", "gcs", "oss", "abfss", "abfs", "wasbs", "wasb")
@@ -1026,16 +1031,12 @@ object CometScanRule extends Logging {
           val residual = residualMethod.invoke(task)
           if (unboundPredicateClass.isInstance(residual)) {
             val term = termMethod.invoke(residual)
-            try {
-              val transformMethod = term.getClass.getMethod("transform")
-              transformMethod.setAccessible(true)
-              val transform = transformMethod.invoke(term)
-              val transformStr = transform.toString
+            // A term with no transform() is a simple reference, which is fine.
+            IcebergReflection.findMethod(term.getClass, "transform").foreach { transformMethod =>
+              val transformStr = transformMethod.invoke(term).toString
               if (transformStr != IcebergReflection.Transforms.IDENTITY) {
                 nonIdentityTransform = Some(transformStr)
               }
-            } catch {
-              case _: NoSuchMethodException => // No transform = simple reference, OK
             }
           }
         } catch {
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index bb81fd69ca..158d4a5995 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -19,6 +19,7 @@
 
 package org.apache.comet.serde.operator
 
+import java.lang.reflect.Method
 import java.math.BigDecimal
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets.UTF_8
@@ -271,11 +272,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
   private def extractDeleteFilesList(
       task: Any,
       contentFileClass: Class[_],
-      fileScanTaskClass: Class[_]): Seq[OperatorOuterClass.IcebergDeleteFile] = {
+      fileScanTaskClass: Class[_],
+      deleteFileClass: Class[_]): Seq[OperatorOuterClass.IcebergDeleteFile] = {
     try {
-      val deleteFileClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.DELETE_FILE)
       // keyMetadata() is declared on ContentFile; present across all supported Iceberg versions.
-      val keyMetadataMethod = contentFileClass.getMethod("keyMetadata")
+      val keyMetadataMethod = IcebergReflection.getMethod(contentFileClass, "keyMetadata")
 
       val deletes = IcebergReflection.getDeleteFilesFromTask(task, fileScanTaskClass)
 
@@ -292,7 +293,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
 
         val contentType =
           try {
-            val contentMethod = deleteFileClass.getMethod("content")
+            val contentMethod = IcebergReflection.getMethod(deleteFileClass, "content")
             val content = contentMethod.invoke(deleteFile)
             content.toString match {
               case IcebergReflection.ContentTypes.POSITION_DELETES =>
@@ -309,7 +310,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
 
         val specId =
           try {
-            val specIdMethod = deleteFileClass.getMethod("specId")
+            val specIdMethod = IcebergReflection.getMethod(deleteFileClass, "specId")
             specIdMethod.invoke(deleteFile).asInstanceOf[Int]
           } catch {
             case _: Exception => 0
@@ -318,7 +319,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
 
         try {
           val equalityIdsMethod =
-            deleteFileClass.getMethod("equalityFieldIds")
+            IcebergReflection.getMethod(deleteFileClass, "equalityFieldIds")
           val equalityIds = equalityIdsMethod
             .invoke(deleteFile)
             .asInstanceOf[java.util.List[Integer]]
@@ -355,29 +356,28 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       task: Any,
       contentScanTaskClass: Class[_],
       fileScanTaskClass: Class[_],
+      partitionSpecToJson: Option[Method],
       taskBuilder: OperatorOuterClass.IcebergFileScanTask.Builder,
       commonBuilder: OperatorOuterClass.IcebergScanCommon.Builder,
       partitionSpecToPoolIndex: mutable.HashMap[String, Int],
       partitionDataToPoolIndex: mutable.HashMap[String, Int]): Unit = {
     try {
-      val specMethod = fileScanTaskClass.getMethod("spec")
+      val specMethod = IcebergReflection.getMethod(fileScanTaskClass, "spec")
       val spec = specMethod.invoke(task)
 
       if (spec != null) {
         // Get the partition type/schema from the spec. Needed regardless of whether this task
         // ends up value-less below.
-        val partitionTypeMethod = spec.getClass.getMethod("partitionType")
+        val partitionTypeMethod = IcebergReflection.getMethod(spec.getClass, "partitionType")
         val partitionType = partitionTypeMethod.invoke(spec)
-        val fieldsMethod = partitionType.getClass.getMethod("fields")
+        val fieldsMethod = IcebergReflection.getMethod(partitionType.getClass, "fields")
         val fields = fieldsMethod
           .invoke(partitionType)
           .asInstanceOf[java.util.List[_]]
 
         // Helper to get field type string (shared by both type and data serialization)
-        def getFieldType(field: Any): String = {
-          val typeMethod = field.getClass.getMethod("type")
-          typeMethod.invoke(field).toString
-        }
+        def getFieldType(field: Any): String =
+          IcebergReflection.getMethod(field.getClass, "type").invoke(field).toString
 
         // Filter out fields with unknown types (dropped partition fields).
         // Unknown type fields represent partition columns that have been dropped
@@ -392,13 +392,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
           if (fieldTypeStr == IcebergReflection.TypeNames.UNKNOWN) {
             None
           } else {
-            val fieldIdMethod = field.getClass.getMethod("fieldId")
+            val fieldIdMethod = IcebergReflection.getMethod(field.getClass, "fieldId")
             val fieldId = fieldIdMethod.invoke(field).asInstanceOf[Int]
 
-            val nameMethod = field.getClass.getMethod("name")
+            val nameMethod = IcebergReflection.getMethod(field.getClass, "name")
             val fieldName = nameMethod.invoke(field).asInstanceOf[String]
 
-            val isOptionalMethod = field.getClass.getMethod("isOptional")
+            val isOptionalMethod = IcebergReflection.getMethod(field.getClass, "isOptional")
             val isOptional =
               isOptionalMethod.invoke(field).asInstanceOf[Boolean]
             val required = !isOptional
@@ -425,12 +425,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         // adds its type-pool entry in the same block to keep them aligned.
         def serializeRealSpec(): Unit = {
           try {
-            val partitionSpecParserClass =
-              IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER)
-            val toJsonMethod = partitionSpecParserClass.getMethod(
-              "toJson",
-              IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC))
-            val partitionSpecJson = toJsonMethod
+            val partitionSpecJson = partitionSpecToJson
+              .getOrElse(throw new NoSuchMethodException("PartitionSpecParser.toJson"))
               .invoke(null, spec)
               .asInstanceOf[String]
 
@@ -457,7 +453,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         }
 
         // Get partition data from the task (via file().partition())
-        val partitionMethod = contentScanTaskClass.getMethod("partition")
+        val partitionMethod = IcebergReflection.getMethod(contentScanTaskClass, "partition")
         val partitionData = partitionMethod.invoke(task)
 
         if (partitionData != null) {
@@ -480,12 +476,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                 None
               } else {
                 // Use the partition type's field ID (same as in partition_type_json)
-                val fieldIdMethod = field.getClass.getMethod("fieldId")
+                val fieldIdMethod = IcebergReflection.getMethod(field.getClass, "fieldId")
                 val fieldId = fieldIdMethod.invoke(field).asInstanceOf[Int]
 
-                val getMethod =
-                  partitionData.getClass.getMethod("get", classOf[Int], classOf[Class[_]])
-                val value = getMethod.invoke(partitionData, Integer.valueOf(idx), classOf[Object])
+                val getValueMethod = IcebergReflection
+                  .getMethod(partitionData.getClass, "get", classOf[Int], classOf[Class[_]])
+                val value =
+                  getValueMethod.invoke(partitionData, Integer.valueOf(idx), classOf[Object])
 
                 Some(partitionValueToProto(fieldId, fieldTypeStr, value))
               }
@@ -506,7 +503,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
           // Only the value-carrying path serializes the real spec (serializeRealSpec), so the
           // value-less path never does the reflection/intern work just to overwrite it.
           if (partitionValues.isEmpty) {
-            val specId = spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int]
+            val specId =
+              IcebergReflection.getMethod(spec.getClass, "specId").invoke(spec).asInstanceOf[Int]
             val emptySpecJson =
               compact(
                 render(("spec-id" -> specId) ~ ("fields" -> List.empty[org.json4s.JObject])))
@@ -630,10 +628,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       val attributeMap = output.map(attr => attr.name -> attr).toMap
 
       if (exprClass.getName.endsWith(Constants.ExpressionTypes.UNBOUND_PREDICATE)) {
-        val operation = exprClass.getMethod("op").invoke(icebergExpr).toString
-        val term = exprClass.getMethod("term").invoke(icebergExpr)
-        val ref = term.getClass.getMethod("ref").invoke(term)
-        val columnName = ref.getClass.getMethod("name").invoke(ref).asInstanceOf[String]
+        val operation = IcebergReflection.getMethod(exprClass, "op").invoke(icebergExpr).toString
+        val term = IcebergReflection.getMethod(exprClass, "term").invoke(icebergExpr)
+        val ref = IcebergReflection.getMethod(term.getClass, "ref").invoke(term)
+        val columnName =
+          IcebergReflection.getMethod(ref.getClass, "name").invoke(ref).asInstanceOf[String]
 
         // Iceberg names a nested reference by its dotted path ("struct.field"), which never matches
         // a top-level scan output attribute, so a residual on a nested field drops here. That miss
@@ -660,11 +659,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         }
       } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.AND)) {
         val left = icebergExprToProto(
-          exprClass.getMethod("left").invoke(icebergExpr),
+          IcebergReflection.getMethod(exprClass, "left").invoke(icebergExpr),
           output,
           pageIndexUnsupportedColumns)
         val right = icebergExprToProto(
-          exprClass.getMethod("right").invoke(icebergExpr),
+          IcebergReflection.getMethod(exprClass, "right").invoke(icebergExpr),
           output,
           pageIndexUnsupportedColumns)
         (left, right) match {
@@ -677,11 +676,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         }
       } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.OR)) {
         val left = icebergExprToProto(
-          exprClass.getMethod("left").invoke(icebergExpr),
+          IcebergReflection.getMethod(exprClass, "left").invoke(icebergExpr),
           output,
           pageIndexUnsupportedColumns)
         val right = icebergExprToProto(
-          exprClass.getMethod("right").invoke(icebergExpr),
+          IcebergReflection.getMethod(exprClass, "right").invoke(icebergExpr),
           output,
           pageIndexUnsupportedColumns)
         // Dropping a disjunct would strengthen the predicate and wrongly prune, so require both.
@@ -690,7 +689,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
           case _ => None
         }
       } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.NOT)) {
-        val child = exprClass.getMethod("child").invoke(icebergExpr)
+        val child = IcebergReflection.getMethod(exprClass, "child").invoke(icebergExpr)
         icebergExprToProto(child, output, pageIndexUnsupportedColumns).map(notPredicate)
       } else {
         None
@@ -722,7 +721,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       attribute: Attribute,
       op: OperatorOuterClass.IcebergPredicateOperator)
       : Option[OperatorOuterClass.IcebergPredicate] = {
-    val literal = exprClass.getMethod("literal").invoke(icebergExpr)
+    val literal = IcebergReflection.getMethod(exprClass, "literal").invoke(icebergExpr)
     predicateLiteralToProto(attribute.dataType, icebergLiteralValue(literal)).map { lit =>
       OperatorOuterClass.IcebergPredicate
         .newBuilder()
@@ -742,7 +741,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       column: String,
       attribute: Attribute): Option[OperatorOuterClass.IcebergPredicate] = {
     val literals =
-      exprClass.getMethod("literals").invoke(icebergExpr).asInstanceOf[java.util.List[_]]
+      IcebergReflection
+        .getMethod(exprClass, "literals")
+        .invoke(icebergExpr)
+        .asInstanceOf[java.util.List[_]]
     val protoLiterals =
       literals.asScala.map(l =>
         predicateLiteralToProto(attribute.dataType, icebergLiteralValue(l)))
@@ -784,7 +786,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
   /** Extracts the raw Java value from an Iceberg Literal via reflection. */
   private def icebergLiteralValue(icebergLiteral: Any): Any = {
     val literalClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.LITERAL)
-    literalClass.getMethod("value").invoke(icebergLiteral)
+    IcebergReflection.getMethod(literalClass, "value").invoke(icebergLiteral)
   }
 
   /**
@@ -906,6 +908,20 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     val deleteFilesToPoolIndex =
       mutable.HashMap[Seq[Int], Int]()
     val residualToPoolIndex = mutable.HashMap[OperatorOuterClass.IcebergPredicate, Int]()
+    // Field-id mappings are read out of an Iceberg schema by reflection, one lookup per column, so
+    // memoize them. Keyed like schemaToPoolIndex above: a task schema that Iceberg materializes
+    // fresh per task misses, but the table/scan schema shared by every task hits.
+    val fieldIdMappingCache = mutable.HashMap[AnyRef, Map[String, Int]]()
+    def fieldIdMapping(schema: AnyRef): Map[String, Int] =
+      fieldIdMappingCache.getOrElseUpdate(schema, IcebergReflection.buildFieldIdMapping(schema))
+    // Whether the scan schema references field ids the current table schema no longer has (a
+    // dropped column read through VERSION AS OF). Loop-invariant, and lazy so a scan whose tasks
+    // all carry deletes never walks the table schema at all.
+    lazy val hasHistoricalColumns = {
+      val tableSchemaFieldIds =
+        fieldIdMapping(metadata.tableSchema.asInstanceOf[AnyRef]).values.toSet
+      metadata.globalFieldIdMapping.values.exists(id => !tableSchemaFieldIds.contains(id))
+    }
     // Columns whose Iceberg type iceberg-rust cannot use for page-index pruning; residual
     // predicates over them are dropped (see icebergExprToProto). Computed once from the full table
     // schema so a filter column projected out of the scan output is still recognized.
@@ -944,19 +960,34 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       IcebergReflection.loadClass(IcebergReflection.ClassNames.SCHEMA_PARSER)
     val schemaClass =
       IcebergReflection.loadClass(IcebergReflection.ClassNames.SCHEMA)
+    val deleteFileClass =
+      IcebergReflection.loadClass(IcebergReflection.ClassNames.DELETE_FILE)
+    // Optional rather than required: serializePartitionData reports an unresolvable
+    // PartitionSpecParser.toJson as a per-task warning that leaves the task without a partition
+    // spec, so it must not fail the whole scan here.
+    val partitionSpecToJson =
+      try {
+        Some(
+          IcebergReflection.getMethod(
+            IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER),
+            "toJson",
+            IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC)))
+      } catch {
+        case _: Exception => None
+      }
 
-    // Cache method lookups (avoid repeated getMethod in loop)
-    val fileMethod = contentScanTaskClass.getMethod("file")
-    val startMethod = contentScanTaskClass.getMethod("start")
-    val lengthMethod = contentScanTaskClass.getMethod("length")
-    val residualMethod = contentScanTaskClass.getMethod("residual")
-    val fileSizeInBytesMethod = contentFileClass.getMethod("fileSizeInBytes")
+    // Accessors used by the per-task loop
+    val fileMethod = IcebergReflection.getMethod(contentScanTaskClass, "file")
+    val startMethod = IcebergReflection.getMethod(contentScanTaskClass, "start")
+    val lengthMethod = IcebergReflection.getMethod(contentScanTaskClass, "length")
+    val residualMethod = IcebergReflection.getMethod(contentScanTaskClass, "residual")
+    val fileSizeInBytesMethod = IcebergReflection.getMethod(contentFileClass, "fileSizeInBytes")
     // keyMetadata() is declared on ContentFile (present across all supported Iceberg versions).
     // For encrypted tables it returns the plaintext StandardKeyMetadata blob; null otherwise.
-    val keyMetadataMethod = contentFileClass.getMethod("keyMetadata")
-    val taskSchemaMethod = fileScanTaskClass.getMethod("schema")
-    val toJsonMethod = schemaParserClass.getMethod("toJson", schemaClass)
-    toJsonMethod.setAccessible(true)
+    val keyMetadataMethod = IcebergReflection.getMethod(contentFileClass, "keyMetadata")
+    val taskSchemaMethod = IcebergReflection.getMethod(fileScanTaskClass, "schema")
+    val toJsonMethod =
+      IcebergReflection.getMethod(schemaParserClass, "toJson", schemaClass)
 
     // Access inputRDD - safe now, DPP is resolved
     scanExec.inputRDD match {
@@ -972,12 +1003,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
             val inputPartClass = inputPartition.getClass
 
             {
-              val taskGroupMethod = inputPartClass.getDeclaredMethod("taskGroup")
-              taskGroupMethod.setAccessible(true)
+              val taskGroupMethod =
+                IcebergReflection.getDeclaredMethod(inputPartClass, "taskGroup")
               val taskGroup = taskGroupMethod.invoke(inputPartition)
 
-              val taskGroupClass = taskGroup.getClass
-              val tasksMethod = taskGroupClass.getMethod("tasks")
+              val tasksMethod = IcebergReflection.getMethod(taskGroup.getClass, "tasks")
               val tasksCollection =
                 tasksMethod.invoke(taskGroup).asInstanceOf[java.util.Collection[_]]
 
@@ -1030,7 +1060,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                     // DeleteFilter.fileProjection).
                     val equalityFieldIds = deletes.asScala.flatMap { df =>
                       IcebergReflection
-                        .getEqualityFieldIds(df)
+                        .getEqualityFieldIds(deleteFileClass, df)
                         .asScala
                         .map(_.asInstanceOf[java.lang.Integer].intValue())
                     }.toSeq
@@ -1042,17 +1072,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                       taskSchema
                     }
                   } else {
-                    val scanSchemaFieldIds = IcebergReflection
-                      .buildFieldIdMapping(metadata.scanSchema)
-                      .values
-                      .toSet
-                    val tableSchemaFieldIds = IcebergReflection
-                      .buildFieldIdMapping(metadata.tableSchema)
-                      .values
-                      .toSet
-                    val hasHistoricalColumns =
-                      scanSchemaFieldIds.exists(id => !tableSchemaFieldIds.contains(id))
-
                     if (hasHistoricalColumns) {
                       metadata.scanSchema.asInstanceOf[AnyRef]
                     } else {
@@ -1069,7 +1088,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                   })
                 taskBuilder.setSchemaIdx(schemaIdx)
 
-                val nameToFieldId = IcebergReflection.buildFieldIdMapping(schema)
+                val nameToFieldId = fieldIdMapping(schema)
 
                 val projectFieldIds = output.map { attr =>
                   nameToFieldId
@@ -1095,7 +1114,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                 taskBuilder.setProjectFieldIdsIdx(projectFieldIdsIdx)
 
                 val deleteFilesList =
-                  extractDeleteFilesList(task, contentFileClass, fileScanTaskClass)
+                  extractDeleteFilesList(
+                    task,
+                    contentFileClass,
+                    fileScanTaskClass,
+                    deleteFileClass)
                 if (deleteFilesList.nonEmpty) {
                   // Intern each delete file into the flat pool, then dedup this task's set as the
                   // resulting list of pool indices.
@@ -1146,6 +1169,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                   task,
                   contentScanTaskClass,
                   fileScanTaskClass,
+                  partitionSpecToJson,
                   taskBuilder,
                   commonBuilder,
                   partitionSpecToPoolIndex,
diff --git a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
index 2ec5f5ac73..64657bffcb 100644
--- a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
@@ -19,12 +19,15 @@
 
 package org.apache.comet.iceberg
 
+import java.lang.reflect.Modifier
 import java.util.Collections
 
 import org.scalatest.funsuite.AnyFunSuite
 
 import org.apache.iceberg.BaseMetastoreTableOperations
 import org.apache.iceberg.BaseTable
+import org.apache.iceberg.DataFiles
+import org.apache.iceberg.PartitionSpec
 import org.apache.iceberg.Schema
 import org.apache.iceberg.TableMetadata
 import org.apache.iceberg.io.FileIO
@@ -44,7 +47,7 @@ class IcebergReflectionSuite extends AnyFunSuite {
     val schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()))
     val expectedMetadata = TableMetadata.newTableMetadata(
       schema,
-      org.apache.iceberg.PartitionSpec.unpartitioned(),
+      PartitionSpec.unpartitioned(),
       "file:///tmp/test-table",
       Collections.emptyMap[String, String]())
     val metadataField = classOf[BaseMetastoreTableOperations]
@@ -63,4 +66,97 @@ class IcebergReflectionSuite extends AnyFunSuite {
     assert(metadata.isDefined)
     assert(metadata.get.isInstanceOf[TableMetadata])
   }
+
+  test("findMethod resolves a method once and returns the cached instance") {
+    val first = IcebergReflection.findMethod(classOf[Schema], "columns")
+    val second = IcebergReflection.findMethod(classOf[Schema], "columns")
+    assert(first.isDefined)
+    assert(first.get.getName == "columns")
+    // Class.getMethod hands back a fresh copy per call; the cache must not.
+    assert(first.get eq second.get)
+  }
+
+  test("an absent method is a cached miss, and getMethod still throws for it") {
+    assert(IcebergReflection.findMethod(classOf[Schema], "noSuchAccessor").isEmpty)
+    assert(IcebergReflection.findMethod(classOf[Schema], "noSuchAccessor").isEmpty)
+    assertThrows[NoSuchMethodException] {
+      IcebergReflection.getMethod(classOf[Schema], "noSuchAccessor")
+    }
+  }
+
+  test("findMethod distinguishes overloads by parameter type") {
+    val byId = IcebergReflection.findMethod(classOf[Schema], "findField", classOf[Int])
+    val byName = IcebergReflection.findMethod(classOf[Schema], "findField", classOf[String])
+    assert(byId.isDefined && byName.isDefined)
+    assert(byId.get ne byName.get)
+
+    val schema = new Schema(Types.NestedField.required(7, "id", Types.IntegerType.get()))
+    val fieldById = byId.get.invoke(schema, Integer.valueOf(7)).asInstanceOf[Types.NestedField]
+    val fieldByName = byName.get.invoke(schema, "id").asInstanceOf[Types.NestedField]
+    assert(fieldById.name() == "id")
+    assert(fieldByName.fieldId() == 7)
+  }
+
+  test("findMethodInHierarchy finds an inherited method and caches it") {
+    val first = IcebergReflection.findMethodInHierarchy(classOf[StubTableOperations], "current")
+    val second = IcebergReflection.findMethodInHierarchy(classOf[StubTableOperations], "current")
+    assert(first.isDefined)
+    // current() is declared on BaseMetastoreTableOperations, not on the stub itself.
+    assert(first.get.getDeclaringClass == classOf[BaseMetastoreTableOperations])
+    assert(first.get eq second.get)
+    assert(IcebergReflection.findMethodInHierarchy(classOf[StubTableOperations], "nope").isEmpty)
+  }
+
+  test("extractFileLocation reads location() when the class has one") {
+    val file = new LocationFile("s3://bucket/data/f.parquet")
+    assert(
+      IcebergReflection.extractFileLocation(classOf[LocationFile], file) ==
+        Some("s3://bucket/data/f.parquet"))
+  }
+
+  test("extractFileLocation falls back to path() on Iceberg versions without location()") {
+    val file = new PathOnlyFile("s3://bucket/data/f.parquet")
+    // Called twice: the second call reads the cached "location() is absent" answer.
+    assert(
+      IcebergReflection.extractFileLocation(classOf[PathOnlyFile], file) ==
+        Some("s3://bucket/data/f.parquet"))
+    assert(
+      IcebergReflection.extractFileLocation(classOf[PathOnlyFile], file) ==
+        Some("s3://bucket/data/f.parquet"))
+  }
+
+  test("extractFileLocation returns None when the class exposes neither accessor") {
+    assert(IcebergReflection.extractFileLocation(classOf[Object], new Object).isEmpty)
+  }
+
+  test("a resolved method has access checks suppressed") {
+    // Iceberg's concrete file impls are package-private (a built DataFile is a GenericDataFile,
+    // and its accessors are declared on the equally package-private BaseFile), so an accessor
+    // resolved on one is not invocable from Comet's package until setAccessible has run. The
+    // modifier assertions keep the test from going vacuous if Iceberg ever makes them public.
+    val file = DataFiles
+      .builder(PartitionSpec.unpartitioned())
+      .withPath("/tmp/data/f.parquet")
+      .withFileSizeInBytes(10)
+      .withRecordCount(1)
+      .withFormat("PARQUET")
+      .build()
+    assert(!Modifier.isPublic(file.getClass.getModifiers))
+
+    val method = IcebergReflection.findMethod(file.getClass, "path")
+    assert(method.isDefined)
+    assert(!Modifier.isPublic(method.get.getDeclaringClass.getModifiers))
+    // Without makeAccessible this invoke throws IllegalAccessException.
+    assert(method.get.invoke(file).toString == "/tmp/data/f.parquet")
+  }
+
+  /** Mimics a newer Iceberg ContentFile, which exposes location(). */
+  class LocationFile(loc: String) {
+    def location(): String = loc
+  }
+
+  /** Mimics Iceberg before 1.7, where ContentFile only exposed path(): CharSequence. */
+  class PathOnlyFile(p: String) {
+    def path(): CharSequence = p
+  }
 }