mbutrovich commented on code in PR #5515: URL: https://github.com/apache/datafusion-comet/pull/5515#discussion_r3982090906
########## spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergDeleteFileSerdeSuite.scala: ########## @@ -0,0 +1,192 @@ +/* + * 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 + +import org.apache.iceberg.DeleteFile + +import org.apache.comet.iceberg.IcebergReflection + +/** + * 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 lookup or 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("required Iceberg delete-file accessors are present") { Review Comment: This is the test that closes the version-bump gap, good addition. Could it assert the values too, not only that the methods exist? ```scala assert(FileContent.POSITION_DELETES.toString == IcebergReflection.ContentTypes.POSITION_DELETES) assert(FileContent.EQUALITY_DELETES.toString == IcebergReflection.ContentTypes.EQUALITY_DELETES) ``` `serializeDeleteFile` now sends `content().toString` straight into the proto and `planner.rs:4364` matches those two literals, so the string values are as load-bearing as the accessors are. The stubs all declare `def content(): String`, so nothing in the suite touches the real enum. `org.apache.iceberg.FileContent` is on the test classpath already, same jar as the `DeleteFile` import on line 26. ########## spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala: ########## @@ -346,6 +306,56 @@ 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 Review Comment: Dropping the match is right, the real validation is already on the native side at `planner.rs:4364`. One consequence worth covering though. `contentType` is now a raw pass-through of the enum's `name()`, and `planner.rs` matches those strings exactly. I checked `iceberg-spark-runtime-4.1_2.13-1.11.0` with javap. `ContentFile.content()` returns `FileContent`, and `FileContent` declares `POSITION_DELETES` and `EQUALITY_DELETES` without overriding `toString`, so the names line up today. Nothing in the suite asserts that. A constant rename in a future Iceberg would pass the new accessor-existence test and then fail every query that has delete files. Separately, after this hunk `IcebergReflection.ContentTypes.POSITION_DELETES` has no callers left. A grep over `spark/`, `common/` and `native/` only finds the three occurrences this hunk deletes, and line 345 is the last use of `EQUALITY_DELETES`. Pinning the names in a test (see my comment on the suite) would give the constant a use again. Otherwise it should probably be removed. ########## spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala: ########## @@ -346,6 +306,56 @@ 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 + deleteBuilder.setContentType(contentType) + + val specIdMethod = IcebergReflection.getMethod(deleteFileClass, "specId") + deleteBuilder.setPartitionSpecId(specIdMethod.invoke(deleteFile).asInstanceOf[Int]) + + val equalityFieldIds = requiredEqualityFieldIds(deleteFileClass, deleteFile) + val isEqualityDelete = + contentType == IcebergReflection.ContentTypes.EQUALITY_DELETES + if (isEqualityDelete && equalityFieldIds.isEmpty) { Review Comment: This is a new fatal case rather than one of the three fallbacks the PR removes. Before, an equality delete with empty ids serialized with no equality ids. Now it fails the query. I think that is correct, an equality delete with no keys cannot be applied, so failing beats guessing. It is not in the description though. The `equalityFieldIds()` bullet currently says `null` is preserved as a legitimate no-equality-keys result. That holds for position deletes only. For an equality delete, `null` and empty are both fatal after this change. Could you add this to the changes list and adjust that bullet? The description becomes the squash commit message on a `Closes #5256` fix, so it is worth having it match what landed. -- 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]
