diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index c7cd0c73424..7f251a2c2d8 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -196,6 +196,7 @@ The following scenarios will fall back to the JVM Iceberg reader: projected is read natively) - Encrypted tables with 192-bit data keys (no AES-192-GCM in the underlying crypto) - Delete files in a format other than Parquet or Puffin (Avro or ORC positional/equality deletes) +- Equality deletes keyed on a field nested in a struct - Tables backed by Avro or ORC data files (only Parquet is accelerated) - Queries that use `input_file_name()`, `input_file_block_start()` or `input_file_block_length()` - Scans whose data or delete files span more than one S3 bucket (the native reader uses one @@ -822,9 +823,6 @@ The physical differences that every native data file has, and why they are safe, ([#7033](https://github.com/apache/datafusion-comet/issues/7033)). - **`VERSION AS OF` reads of a column whose type was promoted after the snapshot** ([#6993](https://github.com/apache/datafusion-comet/issues/6993)). -- **An equality delete keyed on a nested field.** Every query the native scan runs on the table - fails with `field not found`, even one that reads the key - ([#6782](https://github.com/apache/datafusion-comet/issues/6782)). - **Column names that only newer Unicode case data folds together**, such as a pair of Vithkuqi letters, which Unicode 14 added. Queries on such a table fail with a duplicate-field error ([#5540](https://github.com/apache/datafusion-comet/issues/5540)). diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index 03743e0267b..696be0534eb 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -958,6 +958,10 @@ case class CometScanRule(session: SparkSession) try { if (!taskValidation.deleteFiles.isEmpty) { val historicSchemas = IcebergReflection.getAllSchemas(metadata.table) + // Check every schema, as a dropped equality key exists only in the history. + val topLevelIds = (metadata.tableSchema +: historicSchemas) + .flatMap(IcebergReflection.buildFieldIdMapping(_).values) + .toSet val contentFileClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.CONTENT_FILE) val deleteFileClass = @@ -1003,6 +1007,13 @@ case class CometScanRule(session: SparkSession) "schema; resolved from schema history (likely a dropped column)") } fieldInfo match { + // Nested keys fall back: iceberg-rust only looks up top-level columns. + case Some((fieldName, _)) if !topLevelIds(fid) => + hasUnsupportedDeletes = true + fallbackReasons += + s"Equality delete on nested field '$fieldName' (id $fid) is not yet " + + "supported by iceberg-rust, which resolves equality field ids " + + "among top-level columns only." case Some((fieldName, fieldType)) => if (fieldType.contains("struct") || fieldType.equalsIgnoreCase( "variant")) { diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 632dd2df975..e859b23be08 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -1213,6 +1213,65 @@ class CometIcebergNativeSuite } } + // The key `s.k` is nested, which iceberg-rust cannot look up, so the scan must fall back. + test("equality delete keyed on a nested field falls back to Spark") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val table = "test_cat.db.nested_eq_delete" + try { + spark.sql( + s"CREATE TABLE $table (id INT, s STRUCT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '2')") + spark.sql( + s"INSERT INTO $table SELECT CAST(id AS INT), " + + "named_struct('k', CAST(id % 10 AS INT), 'a', CAST(id AS INT)) FROM range(100)") + + val icebergTable = Spark3Util.loadIcebergTable(spark, table) + val deleteRowSchema = icebergTable.schema().select("s.k") + val struct = org.apache.iceberg.data.GenericRecord + .create(deleteRowSchema.findField("s").`type`().asStructType()) + struct.setField("k", Integer.valueOf(3)) + val row = org.apache.iceberg.data.GenericRecord.create(deleteRowSchema) + row.setField("s", struct) + val out = icebergTable + .io() + .newOutputFile( + new File(warehouseDir, s"eq-delete-${System.nanoTime()}.parquet").toString) + val deleteFile = org.apache.iceberg.data.CometEqualityDeletes.write( + icebergTable, + out, + null, + java.util.Collections.singletonList[org.apache.iceberg.data.Record](row), + deleteRowSchema, + Some(Array(icebergTable.schema().findField("s.k").fieldId()))) + icebergTable.newRowDelta().addDeletes(deleteFile).commit() + + // Select the key: if the query leaves it out, even Spark returns the wrong rows. + val df = spark.sql(s"SELECT count(*), sum(s.k), sum(s.a) FROM $table") + checkAnswer(df, Row(90L, 420L, 4470L)) + val plan = df.queryExecution.executedPlan + assert( + collectIcebergNativeScans(plan).isEmpty, + "A nested equality-delete key must prevent the native Iceberg scan") + val fallbackReasons = new ExtendedExplainInfo().getFallbackReasons(plan) + assert( + fallbackReasons.exists(_.contains("Equality delete on nested field 'k'")), + s"Expected nested equality-delete fallback, found: ${fallbackReasons.mkString(", ")}") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + test("MOR table with multiple delete operations - mixed delete types") { assume(icebergAvailable, "Iceberg not available in classpath") diff --git a/spark/src/test/scala/org/apache/iceberg/data/CometEqualityDeletes.scala b/spark/src/test/scala/org/apache/iceberg/data/CometEqualityDeletes.scala index 23cb25e6031..3c77a205271 100644 --- a/spark/src/test/scala/org/apache/iceberg/data/CometEqualityDeletes.scala +++ b/spark/src/test/scala/org/apache/iceberg/data/CometEqualityDeletes.scala @@ -43,12 +43,14 @@ object CometEqualityDeletes { out: OutputFile, partition: StructLike, deletes: java.util.List[Record], - deleteRowSchema: Schema): DeleteFile = { - val equalityFieldIds = deleteRowSchema.columns().asScala.map(_.fieldId()).toArray + deleteRowSchema: Schema, + // Omit to key the delete on every top-level column of `deleteRowSchema`. + equalityFieldIds: Option[Array[Int]] = None): DeleteFile = { val factory = GenericFileWriterFactory .builderFor(table) .equalityDeleteRowSchema(deleteRowSchema) - .equalityFieldIds(equalityFieldIds) + .equalityFieldIds( + equalityFieldIds.getOrElse(deleteRowSchema.columns().asScala.map(_.fieldId()).toArray)) .build() val writer = factory.newEqualityDeleteWriter( EncryptedFiles.encryptedOutput(out, EncryptionKeyMetadata.EMPTY),