Re: [PR] feat: implement native Iceberg V2 writer via iceberg-rust [datafusion-comet]

["andygrove (via GitHub)" <gi...@apache.org>]

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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org