andygrove commented on code in PR #5515:
URL: https://github.com/apache/datafusion-comet/pull/5515#discussion_r3899609342
##########
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:
Nice catch on the `null` return here. I checked
`BaseFile.equalityFieldIds()` and it goes through
`ArrayUtil.toUnmodifiableIntList`, which returns `null` for a null backing
array, so the old `catch` really was swallowing an NPE on every position-delete
file. Without this guard the PR would break every position delete.
One thing though. `IcebergReflection.getEqualityFieldIds` does almost
exactly this, including the null-to-empty mapping, but keeps the blanket catch.
`CometIcebergNativeScan` still calls it a bit further up to decide whether to
union equality-delete field IDs into the task schema via
`schemaWithRequiredFields`. That is the second instance #5256 calls out under
finding 3:
> Finding 3 has a second, independent instance:
`IcebergReflection.getEqualityFieldIds` swallows the same failure into an empty
list, and that result also drives the task-schema decision.
and its Expected behavior section asks for the helper to distinguish a
genuine "no equality ids" from a reflective failure.
Would it work to drop the catch from `getEqualityFieldIds`, keep the
null-to-empty mapping there, and have `serializeDeleteFile` call it? That fixes
both call sites and avoids carrying two copies of this reflection with
different failure semantics.
There is a wrinkle with the third caller. `CometScanRule` calls the same
helper during planning, where falling back to Spark is still an option, so a
blanket change would turn a fallback into a failure there. That is the same
split-entry-point problem #5257 describes for `buildFieldIdMapping`, so it may
be cleaner as a separate change.
If you would rather keep this PR tight, could we switch the description to
"Part of #5256" and file a follow-up for the helper? I checked and it is not
covered by #5257, which enumerates `buildFieldIdMapping`,
`pageIndexUnsupportedColumns` and `PartitionSpecParser.toJson` but not this
one, and #5258 is already closed. So as written the issue would get closed with
half of finding 3 still open.
##########
spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergDeleteFileSerdeSuite.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.serde.operator
+
+import java.lang.reflect.InvocationTargetException
+
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Locks in the fail-loud behavior of
[[CometIcebergNativeScan.serializeDeleteFile]] required by
+ * apache/datafusion-comet#5256: on supported Iceberg versions `content()`,
`specId()`, and
+ * `equalityFieldIds()` are always declared, so a reflective invocation
failure must propagate
+ * rather than fall back to a guessed value. A null `equalityFieldIds()` (a
position-delete file)
+ * stays a legitimate "no equality keys" result.
+ */
+class CometIcebergDeleteFileSerdeSuite extends AnyFunSuite {
+
+ private def keyMetadataMethod(clazz: Class[_]) =
clazz.getMethod("keyMetadata")
+
+ private def serialize(file: AnyRef) =
+ CometIcebergNativeScan.serializeDeleteFile(
+ file,
+ file.getClass,
+ file.getClass,
+ keyMetadataMethod(file.getClass))
+
+ test("position-delete file: null equalityFieldIds() serializes with no
equality ids") {
+ val proto = serialize(new PositionDeleteFile)
+ assert(proto.getContentType == "POSITION_DELETES")
+ assert(proto.getPartitionSpecId == 7)
+ assert(proto.getEqualityIdsCount == 0)
+ assert(proto.getFilePath == "s3://bucket/pos-delete.parquet")
+ }
+
+ test("equality-delete file: declared equalityFieldIds() are serialized") {
+ val proto = serialize(new EqualityDeleteFile)
+ assert(proto.getContentType == "EQUALITY_DELETES")
+ assert(proto.getEqualityIdsCount == 2)
+ assert(proto.getEqualityIds(0) == 3)
+ assert(proto.getEqualityIds(1) == 5)
+ }
+
+ test("content() invocation failure propagates instead of defaulting to
POSITION_DELETES") {
+ val ex = intercept[InvocationTargetException](serialize(new
ThrowingContentDeleteFile))
+ assert(ex.getCause.getMessage == "content boom")
+ }
+
+ test("specId() invocation failure propagates instead of defaulting to 0") {
+ val ex = intercept[InvocationTargetException](serialize(new
ThrowingSpecIdDeleteFile))
+ assert(ex.getCause.getMessage == "spec boom")
+ }
+
+ test("equalityFieldIds() invocation failure propagates instead of dropping
equality ids") {
+ val ex = intercept[InvocationTargetException](serialize(new
ThrowingEqualityIdsDeleteFile))
+ assert(ex.getCause.getMessage == "ids boom")
+ }
+
+ test("missing content() accessor is fatal, not a default") {
+ // getMethod throws NoSuchMethodException directly (not wrapped) when the
accessor is absent.
+ assertThrows[NoSuchMethodException](serialize(new
NoContentAccessorDeleteFile))
+ }
Review Comment:
The synthetic stubs are a clean way to pin the control flow, thanks for that.
The argument for removing the fallbacks is that `content()`, `specId()` and
`equalityFieldIds()` are always declared on the real Iceberg interfaces, and I
do not think anything in the suite checks that. I verified it holds with
`javap` across all four pinned versions, and they are all on `ContentFile`. But
the consequence of that ceasing to be true changed with this PR. It used to be
quietly wrong rows, now it is a hard query failure for anyone with delete files.
That is the regression that matters most now, and it is the one a test would
let CI catch at the version bump rather than in the field. The pieces are
already here, `iceberg-spark-runtime` is a test dependency in every Spark
profile and `IcebergReflectionSuite` already resolves methods against real
Iceberg classes. Would you add something like
`assert(IcebergReflection.findMethod(classOf[DeleteFile],
"content").isDefined)` for each of the three?
Also, the missing-path branch in `serializeDeleteFile` is the one fail-loud
path without a test, and it is the one carrying the longest comment about why
it has to be fatal. A stub declaring neither `location()` nor `path()` would
round the suite out.
##########
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
+ }
Review Comment:
All three branches of this match return their argument unchanged, so the
whole thing is equivalent to `contentMethod.invoke(deleteFile).toString`.
The real validation is already on the native side. `planner.rs:3910` rejects
any content type that is not one of these two with `Invalid delete content type
'{}'`. Since you are rewriting this block anyway, could we drop the match? As
it stands it reads like the Scala side is filtering when it is not.
--
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]
