andygrove commented on code in PR #5361:
URL: https://github.com/apache/datafusion-comet/pull/5361#discussion_r3905939883
##########
docs/source/user-guide/latest/iceberg-writes.md:
##########
@@ -156,23 +170,64 @@ identically on both paths, WAP / branch / snapshot
properties act on the JVM com
settings route the write through `WriteDelta`, which the split plan never
intercepts. Every
rule is pinned by `CometIcebergWriteDetectionSuite`.
-Manifest `DataFile` metrics will be assembled on the JVM at commit time using
Iceberg's own
-`MetricsConfig` logic, so iceberg-java's metadata decisions — metrics modes,
the
-inferred-column cap (`write.metadata.metrics.max-inferred-column-defaults`),
bound truncation,
-and list/map bounds suppression — are respected exactly regardless of what the
native writer
-reports. The `counts`/`none` restrictions above remain only until that
assembly lands.
+Manifest `DataFile` metrics are assembled on the JVM before commit: each
written file's
+metrics are re-derived from its parquet footer through the version-matched
+`ParquetUtil.footerMetrics` and `MetricsConfig.forTable`, with float/double
NaN counts and
+bounds carried over from the native writer's tracked state. iceberg-java's
metadata decisions
+— metrics modes, the inferred-column cap
+(`write.metadata.metrics.max-inferred-column-defaults`), bound truncation, and
list/map bounds
+suppression — are therefore applied by iceberg-java's own code regardless of
what the native
+writer reports. This costs one footer-sized ranged read per written file at
write time.
+
+## Failure handling
+
+Eligibility is decided entirely at plan time. That includes the reflection
surface: every
+iceberg-java class, method, and constructor the executor-side commit-message
assembly uses is
+eagerly resolved by the eligibility gate on the driver, so an Iceberg release
that moves any
+of them declines the native path with a fall-back reason instead of failing
tasks mid-write.
+Once planned, the physical plan is fixed — there is no per-task re-decision or
runtime switch
+back to the JVM writer.
+
+When a native write fails partway through a task (an object-store error, a
data-dependent cast
+failure), the error propagates as an ordinary Spark task failure and Spark's
task retry
+re-executes it — through the native writer again. Retries cannot collide: each
attempt's task
+attempt id is embedded in its data file names.
+
+Partial results are never committed. The commit set is exactly the commit
messages returned by
+successful tasks — a failed task contributes none — and if the job fails, the
driver-side
+commit operator aborts without committing anything. Data files already
finalized by a failed
+task attempt are not deleted by that task (iceberg-java's writer abort deletes
them; the
+native path has no abort hook yet): they are invisible to every reader, since
readers resolve
+files through committed manifests only, and are reclaimed by Iceberg's normal
Review Comment:
This is honest documentation of a real gap, and I appreciate that it's
called out rather than glossed. My worry is that it will not turn into work
unless it gets a tracking issue. In practice this is a per-task-failure orphan
cost that iceberg-java's `SparkPartitionedDataWriter` cleans up synchronously
via its abort hook. On a well-behaved cluster it is invisible; on spot-heavy or
preemption-prone environments it accumulates until someone remembers to
schedule `remove_orphan_files`, and users on the native path may not know
they've opted in to that trade until they hit it.
Could you file a tracking issue for the abort hook (Spark's
`TaskContext.addTaskFailureListener` is the natural spot, mirroring what
`SparkPartitionedDataWriter` does) and link it from this sentence? A pointer
here makes the operational cost visible to users reading the compat page, and
makes it easy for the follow-up to close it out.
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -1110,6 +1268,607 @@ object IcebergReflection extends Logging {
logError(s"Iceberg reflection failure: Failed to get data location:
${e.getMessage}", e)
None
}
+
+ /**
+ * Returns the top-level column names of an Iceberg `Schema`, in declared
order. Used by the
+ * native write serde to project Spark 4.x `ReplaceData` row streams (which
carry an
+ * `__row_operation` column plus optional file-metadata columns) down to
just the data columns
+ * the native iceberg-rust writer expects.
+ */
+ def getSchemaFieldNames(schema: Any): Option[Seq[String]] =
+ try {
+ val cols = schema.getClass
+ .getMethod("columns")
+ .invoke(schema)
+ .asInstanceOf[java.util.List[_]]
+ val names = new scala.collection.mutable.ArrayBuffer[String](cols.size())
+ val it = cols.iterator()
+ while (it.hasNext) {
+ val col = it.next().asInstanceOf[AnyRef]
+ names +=
col.getClass.getMethod("name").invoke(col).asInstanceOf[String]
+ }
+ Some(names.toSeq)
+ } catch {
+ case e: Exception =>
+ logError(s"Iceberg reflection failure: Schema.columns():
${e.getMessage}")
+ None
+ }
+
+ /**
+ * Finds the first field -- nested struct/list/map fields included -- whose
Iceberg type's
+ * `TypeID` name is in `typeIds` (e.g. `Set("UUID")`), returning its `(name,
typeId)`. Used by
+ * write detection to decline schemas containing types the native writer
cannot reproduce.
+ * Reflection failures are deliberately not swallowed: the caller's
detection wrapper turns them
+ * into a fall-back, so a failed walk reads as "cannot verify" rather than
"supported".
+ */
+ def findFieldWithTypeIds(schema: Any, typeIds: Set[String]): Option[(String,
String)] = {
+ val queue = new java.util.ArrayDeque[AnyRef]()
+ queue.addAll(
+
schema.getClass.getMethod("columns").invoke(schema).asInstanceOf[java.util.List[AnyRef]])
+ while (!queue.isEmpty) {
+ val field = queue.poll()
+ val fieldType = field.getClass.getMethod("type").invoke(field)
+ val typeId = fieldType.getClass.getMethod("typeId").invoke(fieldType)
+ val typeIdName = typeId.asInstanceOf[Enum[_]].name()
+ if (typeIds.contains(typeIdName)) {
+ val name =
field.getClass.getMethod("name").invoke(field).asInstanceOf[String]
+ return Some((name, typeIdName))
+ }
+ val isNested =
+
fieldType.getClass.getMethod("isNestedType").invoke(fieldType).asInstanceOf[Boolean]
+ if (isNested) {
+ val nested =
fieldType.getClass.getMethod("asNestedType").invoke(fieldType)
+ queue.addAll(
+ nested.getClass
+ .getMethod("fields")
+ .invoke(nested)
+ .asInstanceOf[java.util.List[AnyRef]])
+ }
+ }
+ None
+ }
+
+ /**
+ * Sum `recordCount` and `fileSizeInBytes` across `dataFiles` for SQL-metric
reporting. The
+ * concrete `DataFile` impl (`BaseFile`) is package-private in Iceberg, so
look the accessors up
+ * on the public `DataFile` interface instead; virtual dispatch still hits
the concrete
+ * implementation at invoke time.
+ */
+ def sumDataFileMetrics(dataFiles: java.util.List[_]): (Long, Long) = {
+ if (dataFiles.isEmpty) return (0L, 0L)
+ val dataFileClass = loadClass(ClassNames.DATA_FILE)
+ val recordCountMethod = dataFileClass.getMethod("recordCount")
+ val fileSizeMethod = dataFileClass.getMethod("fileSizeInBytes")
+ var rows = 0L
+ var bytes = 0L
+ val it = dataFiles.iterator()
+ while (it.hasNext) {
+ val df = it.next().asInstanceOf[AnyRef]
+ rows +=
recordCountMethod.invoke(df).asInstanceOf[java.lang.Long].longValue()
+ bytes +=
fileSizeMethod.invoke(df).asInstanceOf[java.lang.Long].longValue()
+ }
+ (rows, bytes)
+ }
+
+ /**
+ * Looks up a `SortOrder` from `Table.sortOrders()` by its id. Used to
recover the sort order
+ * the write was planned against (`outputSortOrderId`) so it can be
re-applied to the decoded
+ * `DataFile`s through the public `DataFiles.Builder.withSortOrder` --
iceberg-rust's writer
+ * doesn't expose the field, so the manifest comes back with `sort_order_id`
unset. `SortOrder`
+ * is `Serializable`, so the result can ship in a task closure.
+ *
+ * Id 0 falls back to `SortOrder.unsorted()` when absent from the map: a
write whose
+ * `outputSortOrderId` resolves to unsorted (no ordering required) may run
against a table whose
+ * metadata only records its non-trivial sort orders.
+ */
+ def getSortOrderById(table: Any, sortOrderId: Int): Option[AnyRef] =
+ try {
+ val method = table.getClass.getMethod("sortOrders")
+ val orders =
method.invoke(table).asInstanceOf[java.util.Map[java.lang.Integer, AnyRef]]
+ Option(orders.get(java.lang.Integer.valueOf(sortOrderId))).orElse {
+ if (sortOrderId == 0) {
+ val sortOrderClass = loadClass("org.apache.iceberg.SortOrder")
+ Some(sortOrderClass.getMethod("unsorted").invoke(null))
+ } else {
+ None
+ }
+ }
+ } catch {
+ case e: Exception =>
+ logError(
+ "Iceberg reflection failure: Failed to look up sort order " +
+ s"$sortOrderId: ${e.getMessage}")
+ None
+ }
+
+ /**
+ * Construct a `SparkWrite$TaskCommit(DataFile[])` instance for the native
commit path. The
+ * constructor is package-private; `setAccessible(true)` is required on
every Iceberg version.
+ */
+ def buildTaskCommit(dataFiles: java.util.List[_]): AnyRef = {
+ val taskCommitClass =
loadClass("org.apache.iceberg.spark.source.SparkWrite$TaskCommit")
+ val dataFileClass = loadClass("org.apache.iceberg.DataFile")
+ val arrayClass = java.lang.reflect.Array.newInstance(dataFileClass,
0).getClass
+ val ctor = taskCommitClass.getDeclaredConstructor(arrayClass)
+ ctor.setAccessible(true)
+ val array = java.lang.reflect.Array.newInstance(dataFileClass,
dataFiles.size())
+ for (i <- 0 until dataFiles.size()) {
+ java.lang.reflect.Array.set(array, i, dataFiles.get(i))
+ }
+ ctor.newInstance(array.asInstanceOf[AnyRef]).asInstanceOf[AnyRef]
+ }
+
+ /**
+ * Eagerly resolves every class, method, and constructor the executor-side
commit-message
+ * assembly reflects on (`decodeManifestToDataFiles`,
`rebuildDataFilesWithJavaMetrics`,
+ * `sumDataFileMetrics`, `buildTaskCommit`). That code runs after
iceberg-rust has already
+ * written the task's data files, so a reflection miss there is a task
failure; probing the full
+ * surface from the eligibility gate turns an Iceberg release that moves any
of it into a
+ * plan-time fallback instead. Memoized: the linked Iceberg cannot change
within a JVM, and
Review Comment:
Small thing on the memoization scope. `lazy val` on a Scala object caches
per JVM, not per Spark session. The comment says "the linked Iceberg cannot
change within a JVM," and for a long-running driver that is *almost* always
true — but `SparkSession.sql("ADD JAR /path/to/iceberg-1.9.jar")` and
`spark-shell --jars`-style dynamic classpath changes can shift what
`loadClass(...)` resolves to mid-run, and a memoized `Some("...")` from the
first probe would then persist even after the classpath was fixed. The reverse
is worse: a first probe that resolved successfully against an old Iceberg gets
cached as `None`, and a later `ADD JAR` to a moved release would then let
ineligible writes through.
Either the comment should soften the claim ("unless dynamic classpath
changes intervene") or the memoization could be keyed by the resolved Iceberg
class-loader identity, whichever you prefer. This is not a real hazard in the
common deployment shape, but the current comment is more absolute than the
reality.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
