jordepic commented on code in PR #5361: URL: https://github.com/apache/datafusion-comet/pull/5361#discussion_r3906602134
########## 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: Filed https://github.com/apache/datafusion-comet/issues/5618 and linked it from this sentence. The issue spells out the shape you suggested (`TaskContext.addTaskFailureListener`, mirroring `SparkCleanupUtil.deleteTaskFiles`) and splits it into the easy half — failures after the native writer returns, where the decoded `DataFile` paths and the `FileIO` are already in the task closure — and the harder half, failures inside the native write, where closing the gap needs the native operator to report finalized paths alongside the error since iceberg-rust's writers have no abort/Drop cleanup of their own. ########## 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: You're right — and the second direction (a cached `None` letting ineligible writes through after an `ADD JAR` to a moved release) convinced me to fix the memoization rather than soften the comment. The probe result is now keyed by the class loader of the resolved `ContentFile` class: `executorReflectionUnresolved` re-probes whenever `loadClass` resolves Iceberg from a different loader, and the doc comment now states the `ADD JAR` rationale explicitly. Chasing that through also surfaced that the three parquet-mr-bridging lookups (`ParquetUtil.footerMetrics`, `ParquetIO.file`, and the shading-derived `ParquetFileReader.open`) were JVM-lifetime lazy vals, so a re-probe against a new loader could have been satisfied by `Method`s memoized against the old one. They're now resolved together per call (`resolveParquetFooterReflection()`), once per task rather than per file, so both the probe and the executor-side rebuild always reflect the currently-loadable Iceberg. -- 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]
