unikdahal commented on code in PR #5515:
URL: https://github.com/apache/datafusion-comet/pull/5515#discussion_r3907558376
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -346,6 +293,63 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
}
}
+ /**
+ * Serializes a single Iceberg DeleteFile to protobuf.
+ *
+ * `content()`, `specId()`, and `equalityFieldIds()` are declared on the
public `ContentFile` /
+ * `DeleteFile` interfaces across all supported Iceberg versions, so a
`getMethod` miss or an
+ * `invoke` failure on any of them means something is genuinely wrong. None
of the three may
+ * fall back to a default: the scan is already committed to native
execution, and a guessed
+ * content type, partition spec, or dropped equality keys all silently
return wrong rows.
+ * Failures propagate to `extractDeleteFilesList`'s outer catch.
+ */
+ private[operator] def serializeDeleteFile(
+ deleteFile: Any,
+ contentFileClass: Class[_],
+ deleteFileClass: Class[_],
+ keyMetadataMethod: Method): OperatorOuterClass.IcebergDeleteFile = {
+ // The path is the one essential field. A delete file we cannot locate
cannot be applied,
+ // and silently skipping it would leak deleted rows, so treat a missing
path as fatal.
+ val deletePath = IcebergReflection
+ .extractFileLocation(contentFileClass, deleteFile)
+ .getOrElse(
+ throw new RuntimeException(
+ "Neither location() nor path() is declared on this Iceberg version's
" +
+ "ContentFile -- cannot extract delete file path from
FileScanTask"))
+
+ val deleteBuilder = OperatorOuterClass.IcebergDeleteFile.newBuilder()
+ deleteBuilder.setFilePath(deletePath)
+
+ val contentMethod = IcebergReflection.getMethod(deleteFileClass, "content")
+ val contentType = contentMethod.invoke(deleteFile).toString match {
+ case IcebergReflection.ContentTypes.POSITION_DELETES =>
+ IcebergReflection.ContentTypes.POSITION_DELETES
+ case IcebergReflection.ContentTypes.EQUALITY_DELETES =>
+ IcebergReflection.ContentTypes.EQUALITY_DELETES
+ case other => other
+ }
+ deleteBuilder.setContentType(contentType)
+
+ val specIdMethod = IcebergReflection.getMethod(deleteFileClass, "specId")
+
deleteBuilder.setPartitionSpecId(specIdMethod.invoke(deleteFile).asInstanceOf[Int])
+
+ val equalityIdsMethod = IcebergReflection.getMethod(deleteFileClass,
"equalityFieldIds")
+ val equalityIds =
equalityIdsMethod.invoke(deleteFile).asInstanceOf[java.util.List[Integer]]
+ // Iceberg's BaseFile stores equality field IDs in a nullable backing
array, so
+ // equalityFieldIds() returns null for files without equality keys. A null
return is a
+ // normal accessor result, unlike a reflective lookup or invocation
failure, and does not
+ // make serialization fail.
+ if (equalityIds != null) {
+ equalityIds.forEach(id => deleteBuilder.addEqualityIds(id))
+ }
Review Comment:
Good point. I kept the planning helper unchanged since `CometScanRule` still
needs to be able to fall back to Spark.
I added a strict serde-side helper instead and use it in both
`serializeDeleteFile` and the task-schema union path. So lookup/invocation
failures are fatal once native execution is committed, while a null return is
still treated as empty for position deletes.
This should cover the second `#5256` path without changing the planning-time
fallback behavior.
--
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]
