From 4e2cbdb10b1a81e562d96af208518e3bbde4a487 Mon Sep 17 00:00:00 2001 From: unikdahal Date: Sat, 22 Aug 2026 12:58:46 +0530 Subject: [PATCH] Distinguish reflection failure from absence --- .../comet/iceberg/IcebergReflection.scala | 62 ++++++--------- .../operator/CometIcebergNativeScan.scala | 7 +- .../iceberg/IcebergReflectionSuite.scala | 76 +++++++++++++++++++ 3 files changed, 105 insertions(+), 40 deletions(-) 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 078cb61f9da..f375d6bdf9b 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -329,31 +329,22 @@ 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". @@ -361,14 +352,11 @@ object IcebergReflection extends Logging { * `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. 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 158d4a59954..0875ea04a82 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 @@ -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) } 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 64657bffcbf..374b6540c2f 100644 --- a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala +++ b/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") + } }