andygrove commented on code in PR #5652:
URL: https://github.com/apache/datafusion-comet/pull/5652#discussion_r3944281228
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergWriteExec.scala:
##########
@@ -183,6 +186,23 @@ case class CometIcebergWriteExec(
val decoded =
if (manifestBytes.isEmpty) new java.util.ArrayList[AnyRef]()
else IcebergReflection.decodeManifestToDataFiles(manifestBytes, specId)
+ // From here on the JVM knows which files iceberg-rust wrote. If this
task fails before its
+ // commit message reaches the committer (metrics rebuild, `TaskCommit`
construction,
+ // serialization), delete them the way iceberg-java's
`DataWriter.abort()` would; failures
+ // inside the native writer itself are cleaned up on the native side.
+ Option(TaskContext.get()).foreach { tc =>
+ tc.addTaskFailureListener(new TaskFailureListener {
Review Comment:
Agreed, and the "recoverable only by successfully decoding the same manifest
whose decode may fail" framing is the crux — that's the part I'd got wrong.
Cleanup ownership no longer depends on the decode at all.
The native operator now emits the locations it wrote as a second Binary
column next to the manifest, in a trivial framing (a big-endian count, then a
length plus UTF-8 bytes per location) that the JVM walks with a `ByteBuffer`.
`doExecute` registers the failure listener *before* pulling the native payload,
owning nothing at first, and `drainNativePayload` hands it the locations off
that column before it copies the manifest bytes out of the off-heap batch — so
the decode, the metrics rebuild, `TaskCommit`, and serialization are all
covered, and so is the manifest `Array[Byte]` copy itself, which for a large
manifest is where the pressure actually shows up.
The residual window is now just building the location list, which is a few
thousand short strings against the Avro decoder's full `DataFile` objects plus
metrics maps. The decode is strict about consuming the whole column, so a
framing divergence between the two sides fails loudly on every native write
rather than silently handing cleanup a truncated list.
--
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]
