This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 07307b5c53 perf: cache Iceberg reflection lookups on the planning path
(#5222)
07307b5c53 is described below
commit 07307b5c531280ad79ba8dd52feebf3f2fe0f224
Author: Andy Grove <[email protected]>
AuthorDate: Tue Aug 4 15:44:36 2026 -0600
perf: cache Iceberg reflection lookups on the planning path (#5222)
* perf: cache Iceberg reflection lookups on the planning path
`Class.getMethod` linearly scans a class's public methods and returns a
fresh `Method` copy per call, and Comet's Iceberg paths resolve the same
handful of accessors once per file scan task, per partition field and per
delete file. `extractFileLocation` additionally probes for `location()` on
every file, so on Iceberg versions that only have `path()` each call built
a `NoSuchMethodException`.
Resolve lookups through a cache in `IcebergReflection`, keyed on the class
via `ClassValue` so entries die with the class rather than pinning a
classloader. Misses are cached too. Memoize the per-task field-id mapping
and hoist the loop-invariant schema comparison out of the serialization
loop.
* refactor: simplify the Iceberg reflection cache
Fold `setAccessible` into `findMethod` so accessibility is a property of the
resolved method rather than of the call site, dropping the
`findAccessibleMethod` / `getAccessibleMethod` pair and its cache-key
namespace. Share the declared-method resolution between `getDeclaredMethod`
and `findMethodInHierarchy`, and read the cache before `computeIfAbsent` so
a
hit does not allocate a mapping function.
Drop the now-unused single-argument `getFileFormat`, and give
`getEqualityFieldIds` the same class-taking shape so its callers stop
reloading `DeleteFile` per delete file.
In the serde, take the `PartitionSpecParser.toJson` accessor as a plain
`Option[Method]` instead of a by-name parameter backed by a `lazy val`, and
reuse the scan-schema field-id mapping the metadata already carries.
* test: assert access suppression against a real package-private Iceberg
file class
The previous test resolved `transform()` on a suite-local class. scalac
emits
nested classes as public, so the invoke succeeded whether or not
`makeAccessible` had run and only the `isAccessible` flag was really
checked.
Build a real `DataFile` instead: its concrete class `GenericDataFile` and
the
`BaseFile` that declares its accessors are both package-private in every
Iceberg
version Comet builds against, so invoking `path()` from the suite's package
throws `IllegalAccessException` unless the resolved method had access checks
suppressed. Modifier assertions guard against the test going vacuous if
those
classes ever become public.
---
.../apache/comet/iceberg/IcebergReflection.scala | 236 ++++++++++++++-------
.../org/apache/comet/rules/CometScanRule.scala | 35 +--
.../serde/operator/CometIcebergNativeScan.scala | 160 ++++++++------
.../comet/iceberg/IcebergReflectionSuite.scala | 98 ++++++++-
4 files changed, 361 insertions(+), 168 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 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
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
