Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
}
}
Loading