Distinguish reflection failure from absence (#5412)

Author: unikdahal

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala

@@ -329,46 +329,34 @@ object IcebergReflection extends Logging {
    * Different Iceberg versions expose file paths differently:
    *   - Newer versions: location() returns String
    *   - Older versions: path() returns CharSequence
+   *
+   * `None` means neither accessor is declared; a genuine invoke failure propagates instead.
    */
-  def extractFileLocation(contentFileClass: Class[_], file: Any): Option[String] = {
-    try {
-      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 _: Exception => None
+  def extractFileLocation(contentFileClass: Class[_], file: Any): Option[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)
     }
-  }
 
   /**
    * Extracts file location from ContentFile instance using dynamic class lookup.
    */
-  def extractFileLocation(file: Any): Option[String] = {
-    try {
-      val contentFileClass = loadClass(ClassNames.CONTENT_FILE)
-      extractFileLocation(contentFileClass, file)
-    } catch {
-      case _: Exception => None
-    }
-  }
+  def extractFileLocation(file: Any): Option[String] =
+    tryLoadClass(ClassNames.CONTENT_FILE).flatMap(extractFileLocation(_, file))
 
   /**
    * 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.
+   *
+   * `None` means `format()` isn't declared; a genuine invoke failure propagates instead.
    */
-  def getFileFormat(contentFileClass: Class[_], file: Any): Option[String] = {
-    try {
-      findMethod(contentFileClass, "format").map(_.invoke(file).toString)
-    } catch {
-      case _: Exception => None
-    }
-  }
+  def getFileFormat(contentFileClass: Class[_], file: Any): Option[String] =
+    findMethod(contentFileClass, "format").map(_.invoke(file).toString)
 
   /**
    * Gets the Iceberg Table from a SparkScan.
@@ -786,20 +774,18 @@ object IcebergReflection extends Logging {
    *   An Iceberg DeleteFile object
    * @return
    *   List of field IDs used in equality deletes, or empty list for position deletes
+   *
+   * Empty means either `equalityFieldIds()` isn't declared, or it returned `null` (Iceberg's
+   * normal contract for a position-delete file). A genuine invoke failure propagates instead of
+   * collapsing into empty.
    */
-  def getEqualityFieldIds(deleteFileClass: Class[_], deleteFile: Any): java.util.List[_] = {
-    try {
-      val ids =
-        getMethod(deleteFileClass, "equalityFieldIds")
-          .invoke(deleteFile)
-          .asInstanceOf[java.util.List[_]]
-      if (ids == null) new java.util.ArrayList[Any]() else ids
-    } catch {
-      case _: Exception =>
-        // Position delete files return null/empty for equalityFieldIds
-        new java.util.ArrayList[Any]()
+  def getEqualityFieldIds(deleteFileClass: Class[_], deleteFile: Any): java.util.List[_] =
+    findMethod(deleteFileClass, "equalityFieldIds") match {
+      case None => new java.util.ArrayList[Any]()
+      case Some(method) =>
+        val ids = method.invoke(deleteFile).asInstanceOf[java.util.List[_]]
+        if (ids == null) new java.util.ArrayList[Any]() else ids
     }
-  }
 
   /**
    * Gets field name and type from schema by field ID.

spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala

@@ -286,7 +286,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         val deletePath = IcebergReflection
           .extractFileLocation(contentFileClass, deleteFile)
           .getOrElse(
-            throw new RuntimeException("Failed to extract delete file path from FileScanTask"))
+            throw new RuntimeException(
+              "Neither location() nor path() is declared on this Iceberg version's " +
+                "ContentFile -- cannot extract delete file path from FileScanTask"))
 
         val deleteBuilder = OperatorOuterClass.IcebergDeleteFile.newBuilder()
         deleteBuilder.setFilePath(deletePath)
@@ -1026,7 +1028,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                     taskBuilder.setDataFilePath(filePath)
                   case None =>
                     val msg =
-                      "Iceberg reflection failure: Cannot extract file path from data file"
+                      "Neither location() nor path() is declared on this Iceberg version's " +
+                        "ContentFile -- cannot extract file path from data file"
                     logError(msg)
                     throw new RuntimeException(msg)
                 }

spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala

@@ -129,6 +129,54 @@ class IcebergReflectionSuite extends AnyFunSuite {
     assert(IcebergReflection.extractFileLocation(classOf[Object], new Object).isEmpty)
   }
 
+  test("extractFileLocation propagates a genuine invoke failure instead of returning None") {
+    val file = new ThrowingLocationFile
+    val ex = intercept[java.lang.reflect.InvocationTargetException] {
+      IcebergReflection.extractFileLocation(classOf[ThrowingLocationFile], file)
+    }
+    assert(ex.getCause.getMessage == "boom")
+  }
+
+  test("getFileFormat reads format() when declared") {
+    val file = new FormatFile("PARQUET")
+    assert(IcebergReflection.getFileFormat(classOf[FormatFile], file) == Some("PARQUET"))
+  }
+
+  test("getFileFormat returns None when format() is not declared") {
+    assert(IcebergReflection.getFileFormat(classOf[Object], new Object).isEmpty)
+  }
+
+  test("getFileFormat propagates a genuine invoke failure instead of returning None") {
+    val file = new ThrowingFormatFile
+    val ex = intercept[java.lang.reflect.InvocationTargetException] {
+      IcebergReflection.getFileFormat(classOf[ThrowingFormatFile], file)
+    }
+    assert(ex.getCause.getMessage == "boom")
+  }
+
+  test("getEqualityFieldIds reads declared equality field ids") {
+    val ids = java.util.List.of(Integer.valueOf(3), Integer.valueOf(5))
+    val file = new EqualityIdsFile(ids)
+    assert(IcebergReflection.getEqualityFieldIds(classOf[EqualityIdsFile], file) == ids)
+  }
+
+  test("getEqualityFieldIds treats a null return (position delete) as empty, not a failure") {
+    val file = new NullEqualityIdsFile
+    assert(IcebergReflection.getEqualityFieldIds(classOf[NullEqualityIdsFile], file).isEmpty)
+  }
+
+  test("getEqualityFieldIds returns empty when equalityFieldIds() is not declared") {
+    assert(IcebergReflection.getEqualityFieldIds(classOf[Object], new Object).isEmpty)
+  }
+
+  test("getEqualityFieldIds propagates a genuine invoke failure instead of returning empty") {
+    val file = new ThrowingEqualityIdsFile
+    val ex = intercept[java.lang.reflect.InvocationTargetException] {
+      IcebergReflection.getEqualityFieldIds(classOf[ThrowingEqualityIdsFile], file)
+    }
+    assert(ex.getCause.getMessage == "boom")
+  }
+
   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
@@ -159,4 +207,32 @@ class IcebergReflectionSuite extends AnyFunSuite {
   class PathOnlyFile(p: String) {
     def path(): CharSequence = p
   }
+
+  /** location() is declared (not a version difference) but the call itself fails. */
+  class ThrowingLocationFile {
+    def location(): String = throw new RuntimeException("boom")
+  }
+
+  class FormatFile(fmt: String) {
+    def format(): String = fmt
+  }
+
+  /** format() is declared but the call itself fails. */
+  class ThrowingFormatFile {
+    def format(): String = throw new RuntimeException("boom")
+  }
+
+  class EqualityIdsFile(ids: java.util.List[Integer]) {
+    def equalityFieldIds(): java.util.List[Integer] = ids
+  }
+
+  /** Mimics a position-delete file: the accessor is declared and returns null, not a failure. */
+  class NullEqualityIdsFile {
+    def equalityFieldIds(): java.util.List[Integer] = null
+  }
+
+  /** equalityFieldIds() is declared but the call itself fails. */
+  class ThrowingEqualityIdsFile {
+    def equalityFieldIds(): java.util.List[Integer] = throw new RuntimeException("boom")
+  }
 }