From 793e320813da6546fb9b75c448e64e8e806e0b93 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 27 Oct 2026 08:14:43 -0600 Subject: [PATCH 1/2] feat: bump iceberg-rust past #3255 and narrow the nested schema evolution fallback Moves the iceberg-rust pin to b154603a, which matches a data file's nested fields by field id (apache/iceberg-rust#3255). The fallback added by #6543 now covers only a struct whose current fields were all added after a data file was written, the one shape where the native read still differs from Spark. That check runs on the task schema the scan reads. The fallback for nested columns of name-mapped tables (#6790) stays. With the bump, a pruned read of a map value whose projected leaves all differ from the file's physical names fails in iceberg-rust with "Map field must have exactly 2 fields". Closes #6934 --- docs/source/user-guide/latest/iceberg.md | 11 +- native/Cargo.lock | 6 +- native/Cargo.toml | 4 +- .../comet/iceberg/IcebergReflection.scala | 44 ++-- .../apache/comet/rules/CometScanRule.scala | 38 ++-- .../comet/CometIcebergNativeSuite.scala | 214 +++++++++++------- 6 files changed, 189 insertions(+), 128 deletions(-) diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index fd333043013..13f811a68c6 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -217,10 +217,10 @@ The following scenarios will fall back to the JVM Iceberg reader: - Scans with residual filters using `truncate`, `bucket`, `year`, `month`, `day`, or `hour` transform functions (partition pruning still works, but row-level filtering of these transforms falls back) -- Scans that read a struct, array, or map column with a nested field that schema evolution added - or renamed. The native reader cannot yet match such a field to data files written before the - change. The check uses the table's schema history, so the fallback stays after those files are - rewritten +- Scans that read a struct column, or a struct inside an array or map, whose fields were all added + by schema evolution after some data file was written. For such a file the native reader returns a + struct of NULLs or NULL, depending on what else the scan reads, where Spark does not. The check + uses the table's schema history, so the fallback stays after those files are rewritten - Scans that read a struct, array, or map column from a table with `schema.name-mapping.default`, including tables imported with `snapshot`, `migrate`, or `add_files` ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The native reader @@ -801,7 +801,8 @@ Which entries apply depends on which native paths are enabled: ### Reads that differ from Spark - **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such - a file as NULL, and the native scan returns the values stored in the file + a file as NULL, and the native scan returns the values stored in the file. A read of only some + of a struct's fields fails instead ([#6794](https://github.com/apache/datafusion-comet/issues/6794)). - **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as NULL, and two columns that swapped names return each other's values diff --git a/native/Cargo.lock b/native/Cargo.lock index 80741b85406..0c4ba356c32 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -3568,7 +3568,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "aes-gcm", "anyhow", @@ -3626,7 +3626,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "proc-macro2", "quote", @@ -3636,7 +3636,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "anyhow", "async-trait", diff --git a/native/Cargo.toml b/native/Cargo.toml index c858f9e715b..6f3b771e8ee 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -85,8 +85,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false } # Pinned to a revision rather than a release because no published iceberg-rust # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain # version requirement once a release ships on arrow 59. Bump policy: #5645. -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef" } +iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } reqsign-core = "3" [profile.release] diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index bd30f12496e..eb84d2095d0 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -1304,15 +1304,15 @@ object IcebergReflection extends Logging { } /** - * Returns the nested fields of the `fieldIds` columns of `schema` that some schema in the - * table's history (`table.schemas()`) lacks or names differently. Each is a dotted path with - * the change, for example `items.element.z (renamed from a)`. A data file written under that - * older schema lacks the field, or has it under the old name. Fields are matched by id at every - * level (struct fields, list elements, map keys and values), so a reorder or a type promotion - * is not a change, and an older schema that lacks the column itself is skipped. Throws on + * Returns the structs under the `fieldIds` columns of `schema`, at any depth and including the + * columns themselves, that share none of their fields with the same struct in some schema in + * the table's history (`table.schemas()`). Each is a dotted path, for example + * `items.element.s`. A data file written under that older schema has none of the struct's + * current fields. Structs are matched by field id at every level (struct fields, list elements, + * map keys and values), so an older schema that lacks the column itself is skipped. Throws on * reflection failure so the caller can fall back. */ - def nestedFieldsAddedOrRenamed(table: Any, schema: Any, fieldIds: Set[Int]): Seq[String] = { + def structsWithNoFieldsInCommon(table: Any, schema: Any, fieldIds: Set[Int]): Seq[String] = { import scala.jdk.CollectionConverters._ val findField = getMethod(schema.getClass, "findField", classOf[Int]) def fieldById(s: Any, id: Int): Option[Any] = @@ -1324,31 +1324,33 @@ object IcebergReflection extends Logging { .asScala .toSeq fieldIds.toSeq.flatMap { id => - fieldById(schema, id).filter(childFields(_).nonEmpty).toSeq.flatMap { field => + fieldById(schema, id).toSeq.flatMap { field => history.flatMap(fieldById(_, id)).flatMap { older => - childFieldsChangedFrom(field, older, fieldName(field)) + structsWithNoFieldsInCommon(field, older, fieldName(field)) } } }.distinct } - /** The nested fields under `field`, at any depth, that `older` lacks or names differently. */ - private def childFieldsChangedFrom(field: Any, older: Any, path: String): Seq[String] = { + /** `field` if it is a struct sharing no field with `older`, plus the same for its children. */ + private def structsWithNoFieldsInCommon(field: Any, older: Any, path: String): Seq[String] = { + val children = childFields(field) val olderChildren = childFields(older).map(child => fieldIdOf(child) -> child).toMap - childFields(field).flatMap { child => - val name = fieldName(child) - val childPath = s"$path.$name" - olderChildren.get(fieldIdOf(child)) match { - case Some(olderChild) => - val olderName = fieldName(olderChild) - val renamed = - if (olderName != name) Seq(s"$childPath (renamed from $olderName)") else Nil - renamed ++ childFieldsChangedFrom(child, olderChild, childPath) - case None => Seq(s"$childPath (added)") + val disjoint = isStruct(field) && children.nonEmpty && + !children.exists(child => olderChildren.contains(fieldIdOf(child))) + val here = if (disjoint) Seq(path) else Nil + here ++ children.flatMap { child => + olderChildren.get(fieldIdOf(child)).toSeq.flatMap { olderChild => + structsWithNoFieldsInCommon(child, olderChild, s"$path.${fieldName(child)}") } } } + private def isStruct(field: Any): Boolean = { + val fieldType = getMethod(field.getClass, "type").invoke(field) + getMethod(fieldType.getClass, "isStructType").invoke(fieldType).asInstanceOf[Boolean] + } + /** A struct's fields, a list's element, or a map's key and value. Empty for other types. */ private def childFields(field: Any): Seq[Any] = { import scala.jdk.CollectionConverters._ 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 349a9dae380..5df5d2b2c03 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -831,16 +831,17 @@ case class CometScanRule(session: SparkSession) false } - // Neither iceberg-rust nor the batch adaptation after it matches a data file's nested - // fields to the table's by field id (apache/iceberg-rust#2617). A file written before a - // nested field was added (to a struct, or to a struct inside a list or map) fails the - // native scan with "Incorrect number of arrays for StructArray fields", and a nested field - // renamed since the file was written can read back as NULL. This check walks the schema - // the scan reads (see CometIcebergNativeScanMetadata.taskSchema), which for VERSION AS OF - // can be older than the table's, and also the table and scan schemas, so it can fall back - // when the changed field is pruned away. A FileScanTask does not record which schema wrote - // its file, so fall back when any schema in the table's history lacks a nested field of a - // projected column or names it differently. + // iceberg-rust matches a data file's nested fields to the table's by field id, filling in + // a field the file lacks with NULLs and applying renames, so schema evolution that adds, + // renames, drops or reorders nested fields reads correctly. One shape still differs from + // Spark: a struct (at any depth) whose fields in the schema the scan reads were all added + // after a file was written. The native read returns a struct of NULLs or NULL there, + // depending on what else the scan reads, where Spark returns NULL for a top-level struct + // and a struct of NULLs for a nested one. The schema the scan reads is the task schema (see + // CometIcebergNativeScanMetadata.taskSchema), which holds only the nested fields Spark + // keeps while nested schema pruning is on, and which for VERSION AS OF can be older than + // the table's. A FileScanTask does not record which schema wrote its file, so fall back + // when any schema in the table's history has such a struct in a projected column. val nestedFieldsSupported = try { val schemas = Seq(metadata.tableSchema, metadata.scanSchema) ++ metadata.taskSchema @@ -850,18 +851,17 @@ case class CometScanRule(session: SparkSession) } else { resolvedProjectedFieldIds } - val changed = schemas - .flatMap { schema => - IcebergReflection.nestedFieldsAddedOrRenamed(metadata.table, schema, fieldIds) - } + val disjoint = metadata.taskSchema.toSeq + .flatMap(IcebergReflection.structsWithNoFieldsInCommon(metadata.table, _, fieldIds)) .distinct .sorted - if (changed.nonEmpty) { - fallbackReasons += "Nested fields added or renamed by Iceberg schema evolution are " + - "not yet supported by Comet's native reader, which cannot match them to data " + - s"files written before the change: ${changed.mkString(", ")}" + if (disjoint.nonEmpty) { + fallbackReasons += "Structs whose fields were all added by Iceberg schema " + + "evolution are not yet supported by Comet's native reader, which reads data " + + "files written before the change differently from Spark: " + + disjoint.mkString(", ") } - changed.isEmpty + disjoint.isEmpty } catch { case e: Exception => fallbackReasons += "Iceberg reflection failure: could not compare nested fields " + diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 5dfbd19fcd0..8534b7f1761 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -30,7 +30,7 @@ import scala.jdk.CollectionConverters._ import org.apache.iceberg.data.IcebergGenerics import org.apache.iceberg.expressions.Expressions import org.apache.iceberg.spark.Spark3Util -import org.apache.spark.CometListenerBusUtils +import org.apache.spark.{CometListenerBusUtils, SparkException} import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.{Ascending, DynamicPruningExpression, SortOrder} @@ -121,13 +121,24 @@ class CometIcebergNativeSuite } /** - * Verifies query correctness and that it falls back to Spark because a projected column's - * nested fields were added or renamed over the table's schema history, as `fields` lists them - * (for example `s.b (added)`). + * Verifies query correctness and that it reads `scans` native Iceberg scans, for a query that + * reads the same table more than once. */ - private def checkNestedFieldEvolutionFallback(query: String, fields: String): Unit = { + private def checkIcebergNativeScans(query: String, scans: Int): Unit = { + val (_, cometPlan) = checkSparkAnswer(query) + assert( + collectIcebergNativeScans(cometPlan).length == scans, + s"Expected $scans CometIcebergNativeScanExec nodes. Plan:\n$cometPlan") + } + + /** + * Verifies query correctness and that it falls back to Spark because a projected column has + * structs, listed in `structs` by path (for example `items.element`), whose fields were all + * added after some data files were written. + */ + private def checkNestedFieldEvolutionFallback(query: String, structs: String): Unit = { val (_, cometPlan) = - checkSparkAnswerAndFallbackReason(query, s"files written before the change: $fields") + checkSparkAnswerAndFallbackReason(query, s"differently from Spark: $structs") assert( collectIcebergNativeScans(cometPlan).isEmpty, s"Expected fallback to Spark but found a CometIcebergNativeScanExec. Plan:\n$cometPlan") @@ -1861,11 +1872,7 @@ class CometIcebergNativeSuite } } - // The native read does not match a data file's nested fields to the table's by field id - // (apache/iceberg-rust#2617). It fails on a file that lacks a nested field the table has gained, - // and can return NULL for a nested field renamed since the file was written, so these reads - // must fall back. - test("schema evolution - nested field added to a struct falls back") { + test("schema evolution - nested field added to a struct stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -1887,34 +1894,27 @@ class CometIcebergNativeSuite spark.sql(s"ALTER TABLE $table ADD COLUMN s.b INT") spark.sql(s"INSERT INTO $table VALUES (3, named_struct('a', 3, 'b', 30))") - checkNestedFieldEvolutionFallback(s"SELECT id, s FROM $table ORDER BY id", "s.b (added)") + // The first file lacks s.b, which the native read fills with NULLs. + checkIcebergNativeScan(s"SELECT id, s FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT s FROM $table ORDER BY s.a") // Spark pushes these null checks into the scan. - checkNestedFieldEvolutionFallback( - s"SELECT id, s FROM $table WHERE s IS NOT NULL ORDER BY id", - "s.b (added)") - checkNestedFieldEvolutionFallback( - s"SELECT id FROM $table WHERE s IS NULL ORDER BY id", - "s.b (added)") - // The native read asks only for s.a here, but the check also covers the column's full - // type in the table schema, so it still falls back. - checkNestedFieldEvolutionFallback( - s"SELECT id, s.a FROM $table ORDER BY id", - "s.b (added)") - checkNestedFieldEvolutionFallback( + checkIcebergNativeScan(s"SELECT id, s FROM $table WHERE s IS NOT NULL ORDER BY id") + checkIcebergNativeScan(s"SELECT id FROM $table WHERE s IS NULL ORDER BY id") + checkIcebergNativeScan(s"SELECT id, s.a FROM $table ORDER BY id") + checkIcebergNativeScans( s"SELECT l.id, r.id FROM $table l JOIN $table r ON l.s = r.s ORDER BY l.id", - "s.b (added)") - // The old snapshot's files lack s.b too, and the scan's full schema is the current one, - // which has it. - checkNestedFieldEvolutionFallback( - s"SELECT id, s FROM $table VERSION AS OF $snapshotBeforeAdd ORDER BY id", - "s.b (added)") - - // A read that does not project the evolved column stays native. + 2) + // The scan reads only s.b, which the first file lacks, so that file has none of the fields + // the scan reads. + checkNestedFieldEvolutionFallback(s"SELECT id, s.b FROM $table ORDER BY id", "s") + // The old snapshot's files lack s.b too, and its schema does not have the field. + checkIcebergNativeScan(s"SELECT id, s FROM $table VERSION AS OF $snapshotBeforeAdd") checkIcebergNativeScan(s"SELECT id FROM $table ORDER BY id") // A VERSION AS OF read whose full schema is older than the table's: the pruned-away // nested partition source `s.k` makes the scan read its full schema, and the dropped `c` - // makes that the snapshot's, which still has the s.b that the first file predates. + // makes that the snapshot's, which still has the s.b that the first file predates. The + // native read fills it with NULLs. val historical = "test_cat.db.evo_struct_historical" spark.sql(s""" CREATE TABLE $historical (id INT, c INT, s STRUCT) @@ -1928,12 +1928,11 @@ class CometIcebergNativeSuite val snapshotWithB = latestSnapshotId(historical) spark.sql(s"ALTER TABLE $historical DROP COLUMN c") spark.sql(s"ALTER TABLE $historical DROP COLUMN s.b") - checkNestedFieldEvolutionFallback( - s"SELECT id, c, s.a FROM $historical VERSION AS OF $snapshotWithB ORDER BY id", - "s.b (added)") + checkIcebergNativeScan( + s"SELECT id, c, s.a FROM $historical VERSION AS OF $snapshotWithB ORDER BY id") - // Unpartitioned, the same read takes the pruned schema, which leaves s.b out, so it stays - // native. With pruning off it reads the full schema and falls back. + // Unpartitioned, the same read takes the pruned schema, which leaves s.b out. With pruning + // off it reads the full schema. Both stay native. val unpartitioned = "test_cat.db.evo_struct_historical_unpartitioned" spark.sql(s"CREATE TABLE $unpartitioned (id INT, c INT, s STRUCT) USING iceberg") spark.sql(s"INSERT INTO $unpartitioned VALUES (1, 100, named_struct('a', 10))") @@ -1946,7 +1945,7 @@ class CometIcebergNativeSuite s"SELECT id, c, s.a FROM $unpartitioned VERSION AS OF $unpartitionedSnapshot ORDER BY id" checkIcebergNativeScan(unpartitionedQuery) withSQLConf(CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") { - checkNestedFieldEvolutionFallback(unpartitionedQuery, "s.b (added)") + checkIcebergNativeScan(unpartitionedQuery) } // The task schema is chosen during planning, so a query planned with pruning on still // reads the pruned schema when it runs with pruning off. @@ -1961,7 +1960,7 @@ class CometIcebergNativeSuite } } - test("schema evolution - nested field added inside an array or a map falls back") { + test("schema evolution - nested field added inside an array or a map stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -1979,19 +1978,12 @@ class CometIcebergNativeSuite spark.sql(s"ALTER TABLE $listTable ADD COLUMN items.element.b INT") spark.sql(s"INSERT INTO $listTable VALUES (3, array(named_struct('a', 30, 'b', 300)))") - checkNestedFieldEvolutionFallback( - s"SELECT id, items FROM $listTable ORDER BY id", - "items.element.b (added)") + checkIcebergNativeScan(s"SELECT id, items FROM $listTable ORDER BY id") // Spark infers isnotnull(items) below a non-outer generator and pushes it into the scan. - checkNestedFieldEvolutionFallback( - s"SELECT id, e.a, e.b FROM $listTable LATERAL VIEW explode(items) x AS e ORDER BY id", - "items.element.b (added)") - checkNestedFieldEvolutionFallback( - s"SELECT id, posexplode(items) FROM $listTable ORDER BY id", - "items.element.b (added)") - checkNestedFieldEvolutionFallback( - s"SELECT id, inline(items) FROM $listTable ORDER BY id", - "items.element.b (added)") + checkIcebergNativeScan( + s"SELECT id, e.a, e.b FROM $listTable LATERAL VIEW explode(items) x AS e ORDER BY id") + checkIcebergNativeScan(s"SELECT id, posexplode(items) FROM $listTable ORDER BY id") + checkIcebergNativeScan(s"SELECT id, inline(items) FROM $listTable ORDER BY id") val mapTable = "test_cat.db.evo_map" spark.sql(s"CREATE TABLE $mapTable (id INT, m MAP>) USING iceberg") @@ -1999,14 +1991,10 @@ class CometIcebergNativeSuite spark.sql(s"ALTER TABLE $mapTable ADD COLUMN m.value.b INT") spark.sql(s"INSERT INTO $mapTable VALUES (3, map('k', named_struct('a', 3, 'b', 30)))") - checkNestedFieldEvolutionFallback( - s"SELECT id, m FROM $mapTable ORDER BY id", - "m.value.b (added)") - checkNestedFieldEvolutionFallback( - s"SELECT id, explode(m) FROM $mapTable ORDER BY id", - "m.value.b (added)") + checkIcebergNativeScan(s"SELECT id, m FROM $mapTable ORDER BY id") + checkIcebergNativeScan(s"SELECT id, explode(m) FROM $mapTable ORDER BY id") - // A field added several levels down, and a struct-typed field, are found too. + // A field added several levels down, and a struct-typed field. val deepTable = "test_cat.db.evo_deep" spark.sql( s"CREATE TABLE $deepTable (id INT, s STRUCT>>) USING iceberg") @@ -2014,9 +2002,9 @@ class CometIcebergNativeSuite s"INSERT INTO $deepTable VALUES (1, named_struct('l', array(named_struct('a', 1))))") spark.sql(s"ALTER TABLE $deepTable ADD COLUMN s.l.element.b INT") spark.sql(s"ALTER TABLE $deepTable ADD COLUMN s.t STRUCT") - checkNestedFieldEvolutionFallback( - s"SELECT id, s FROM $deepTable ORDER BY id", - "s.l.element.b (added), s.t (added)") + spark.sql( + s"INSERT INTO $deepTable VALUES (2, named_struct('l', array(named_struct('a', 2, 'b', 20)), 't', named_struct('x', 5)))") + checkIcebergNativeScan(s"SELECT id, s FROM $deepTable ORDER BY id") spark.sql(s"DROP TABLE $listTable") spark.sql(s"DROP TABLE $mapTable") @@ -2025,7 +2013,7 @@ class CometIcebergNativeSuite } } - test("schema evolution - nested field dropped and re-added under the same name falls back") { + test("schema evolution - nested field dropped and re-added under the same name stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -2037,9 +2025,9 @@ class CometIcebergNativeSuite CometConf.COMET_EXEC_ENABLED.key -> "true", CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { - // The re-added s.b has a new field id, so the old file's s.b is not it. Drop and re-add it - // in one schema change, so every schema in the history has a field named s.b and only - // the field id tells them apart. + // The re-added s.b has a new field id, so the old file's s.b is not it and reads as NULL. + // Drop and re-add it in one schema change, so every schema in the history has a field + // named s.b and only the field id tells them apart. val table = "test_cat.db.evo_readd" spark.sql(s"CREATE TABLE $table (id INT, s STRUCT) USING iceberg") spark.sql(s"INSERT INTO $table VALUES (1, named_struct('a', 1, 'b', 2))") @@ -2052,14 +2040,14 @@ class CometIcebergNativeSuite spark.sql(s"REFRESH TABLE $table") spark.sql(s"INSERT INTO $table VALUES (2, named_struct('a', 3, 'b', 4))") - checkNestedFieldEvolutionFallback(s"SELECT id, s FROM $table ORDER BY id", "s.b (added)") + checkIcebergNativeScan(s"SELECT id, s FROM $table ORDER BY id") spark.sql(s"DROP TABLE $table") } } } - test("schema evolution - nested field renamed falls back") { + test("schema evolution - nested field renamed stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -2080,8 +2068,7 @@ class CometIcebergNativeSuite moved STRUCT ) USING iceberg """) - // The nulls make Iceberg write the nested fields as optional, as the table declares them, - // which is the layout where the native read returns NULL for a renamed field. + // The nulls make Iceberg write the nested fields as optional, as the table declares them. spark.sql(s""" INSERT INTO $table VALUES (1, named_struct('a', 1, 'b', 'x'), array(named_struct('a', 1, 'b', 2)), @@ -2100,15 +2087,80 @@ class CometIcebergNativeSuite named_struct('b', 4, 'c', 3)) """) + checkIcebergNativeScan(s"SELECT id, s FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT id, items FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT id, moved FROM $table ORDER BY id") + + spark.sql(s"DROP TABLE $table") + } + } + } + + // Spark reads a struct whose current fields all postdate a data file as NULL at the top level and + // as a struct of NULLs inside another struct. The native read gives a struct of NULLs or NULL + // depending on what else the scan reads, so these reads fall back. + test("schema evolution - struct whose fields were all added falls back") { + 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.evo_disjoint" + spark.sql(s""" + CREATE TABLE $table ( + id INT, + s STRUCT, + items ARRAY>, + outer_s STRUCT>, + partial STRUCT + ) USING iceberg + """) + spark.sql(s""" + INSERT INTO $table VALUES + (1, named_struct('a', 1), array(named_struct('a', 1)), + named_struct('x', 1, 'inner_s', named_struct('a', 1)), named_struct('a', 1, 'b', 2)), + (2, null, null, null, null) + """) + spark.sql(s"ALTER TABLE $table ADD COLUMN s.b INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN s.a") + spark.sql(s"ALTER TABLE $table ADD COLUMN items.element.b INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN items.element.a") + spark.sql(s"ALTER TABLE $table ADD COLUMN outer_s.inner_s.b INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN outer_s.inner_s.a") + // partial keeps b, so a file that predates the change still has one of its fields. + spark.sql(s"ALTER TABLE $table ADD COLUMN partial.c INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN partial.a") + spark.sql(s""" + INSERT INTO $table VALUES + (3, named_struct('b', 3), array(named_struct('b', 3)), + named_struct('x', 3, 'inner_s', named_struct('b', 3)), named_struct('b', 3, 'c', 4)) + """) + + checkNestedFieldEvolutionFallback(s"SELECT s FROM $table ORDER BY id", "s") + checkNestedFieldEvolutionFallback(s"SELECT id, s FROM $table ORDER BY id", "s") + checkNestedFieldEvolutionFallback(s"SELECT id, s.b FROM $table ORDER BY id", "s") + checkNestedFieldEvolutionFallback( + s"SELECT id FROM $table WHERE s IS NULL ORDER BY id", + "s") checkNestedFieldEvolutionFallback( - s"SELECT id, s FROM $table ORDER BY id", - "s.z (renamed from a)") + s"SELECT items FROM $table ORDER BY id", + "items.element") checkNestedFieldEvolutionFallback( - s"SELECT id, items FROM $table ORDER BY id", - "items.element.z (renamed from a)") + s"SELECT outer_s FROM $table ORDER BY id", + "outer_s.inner_s") checkNestedFieldEvolutionFallback( - s"SELECT id, moved FROM $table ORDER BY id", - "moved.c (renamed from a)") + s"SELECT id, outer_s, items, s FROM $table ORDER BY id", + "items.element, outer_s.inner_s, s") + + // A read that does not project those columns, or whose struct kept a field, stays native. + checkIcebergNativeScan(s"SELECT id FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT id, partial FROM $table ORDER BY id") spark.sql(s"DROP TABLE $table") } @@ -2908,13 +2960,19 @@ class CometIcebergNativeSuite } rows } - // `s.b` is the struct's last field, so a positional cast would read `s.a` instead. + // Without field ids or a name mapping, iceberg-rust matches a struct's fields to the + // table's by position, and only when their counts are equal. A read of the whole struct + // gets the file's values, as it did before. A pruned read of some of its fields fails + // rather than reading a different field, since `s.b` is the struct's last field. Seq( s"SELECT id, s.b FROM $table ORDER BY id", s"SELECT id, items.x FROM $table ORDER BY id").foreach { query => val full = nativeRows(query, pruneNestedFields = false) assert(full.length == 100 && full.forall(!_.isNullAt(1)), s"$query: $full") - assert(nativeRows(query, pruneNestedFields = true) == full, query) + val e = intercept[SparkException](nativeRows(query, pruneNestedFields = true)) + assert( + e.getMessage.contains("cannot reconcile struct fields by id"), + s"$query: ${e.getMessage}") } spark.sql(s"DROP TABLE $table") From bc89e2cc372c96f8d74a0d84367c87d1423a08b8 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 27 Oct 2026 12:13:24 -0600 Subject: [PATCH 2/2] fix: bind equality deletes case-sensitively and check nested defaults on the task schema iceberg-rust now rejects a case-insensitive name that matches two columns when it binds an equality delete's predicate to the task schema, so a table with `id` and `ID` failed any scan whose task schema held both. The names come from that schema, so Comet marks its tasks case-sensitive. The check for V3 default values read only the query's schema. While nested schema pruning is off, the scan reads every nested field of a projected column, and iceberg-rust fails a nested field with a default that a data file lacks. The check now also covers the projected columns of the task schema. --- native/core/src/execution/planner.rs | 6 +- .../comet/iceberg/IcebergReflection.scala | 18 +++++ .../apache/comet/rules/CometScanRule.scala | 44 +++++++---- .../comet/CometIcebergNativeSuite.scala | 74 +++++++++++++++++++ .../apache/comet/CometIcebergTestBase.scala | 27 ++++--- 5 files changed, 142 insertions(+), 27 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 0241e1f1ce8..c6641943e8f 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -2739,7 +2739,11 @@ fn parse_file_scan_tasks_from_common( .with_partition_spec(partition_spec) .with_name_mapping(name_mapping) .with_unified_partition_type(unified_partition_type_for_task) - .with_case_sensitive(false) + // The only consumer is the binding of an equality delete's predicate, whose + // column names iceberg-rust takes from the task schema. They match it exactly, + // and a case-insensitive binding rejects a schema with two names that differ + // only by case. + .with_case_sensitive(true) // Plaintext StandardKeyMetadata forwarded verbatim from the JVM; decoded by // iceberg-rust with no KMS unwrap. None for unencrypted data files. .with_key_metadata(proto_task.key_metadata.clone().map(Vec::into_boxed_slice)) diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index eb84d2095d0..ff318e0dfb6 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -1283,6 +1283,24 @@ object IcebergReflection extends Logging { columns.asScala.flatMap(walkFieldForDefault).toList } + /** + * [[columnsWithInitialDefault]] for only the top-level columns of `schema` whose ids are in + * `columnIds`. A scan reads every nested field of such a column that its schema holds, so this + * is for the schema the scan reads when that is wider than the query's, as the table's is while + * nested schema pruning is off. + */ + def columnsWithInitialDefault(schema: Any, columnIds: Set[Int]): List[String] = { + import scala.jdk.CollectionConverters._ + val columns = + getMethod(schema.getClass, "columns").invoke(schema).asInstanceOf[java.util.List[_]] + columns.asScala + .filter(column => columnIds.contains(fieldIdOf(column))) + .flatMap { + walkFieldForDefault + } + .toList + } + private def walkFieldForDefault(field: Any): List[String] = { import scala.jdk.CollectionConverters._ val name = getMethod(field.getClass, "name").invoke(field).asInstanceOf[String] 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 5df5d2b2c03..955db40ce42 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -743,17 +743,41 @@ case class CometScanRule(session: SparkSession) false } + // Projected roots are matched by field ID so historical snapshots still identify renamed + // columns. Both schema checks below use them. + val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol) + val resolver = session.sessionState.conf.resolver + val projectedFieldIds = projectedDataColumns.map { attr => + metadata.globalFieldIdMapping.collectFirst { + case (fieldName, fieldId) if resolver(fieldName, attr.name) => fieldId + } + } + val resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet + val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty) + // Column default values are a V3 feature, so V1/V2 tables cannot have them and need no // check. Only inspect V3 tables. iceberg-rust does not synthesize a V3 initial-default for - // a column absent from a data file, so a scan projecting such a column must fall back. A V3 - // table implies an Iceberg version new enough to expose initialDefault(), so a reflection - // failure here is unexpected and also falls back rather than risk a crash. + // a column absent from a data file (it fails a nested one), so a scan reading such a + // column must fall back. That is the query's schema and, while nested schema pruning is + // off, the whole of each projected column in the task schema, which holds nested fields + // the query's does not. A V3 table implies an Iceberg version new enough to expose + // initialDefault(), so a reflection failure here is unexpected and also falls back rather + // than risk a crash. val defaultValuesSupported = if (!formatVersion.exists(_ >= 3)) { true } else { try { - val defaulted = IcebergReflection.columnsWithInitialDefault(metadata.scanSchema) + val readDefaults = metadata.taskSchema.toSeq.flatMap { schema => + if (hasUnresolvedProjectedFieldIds) { + IcebergReflection.columnsWithInitialDefault(schema) + } else { + IcebergReflection.columnsWithInitialDefault(schema, resolvedProjectedFieldIds) + } + } + val defaulted = + (IcebergReflection.columnsWithInitialDefault( + metadata.scanSchema) ++ readDefaults).distinct if (defaulted.nonEmpty) { fallbackReasons += "Iceberg column(s) with V3 default values are not yet " + s"supported by Comet's native reader: ${defaulted.mkString(", ")}" @@ -769,18 +793,6 @@ case class CometScanRule(session: SparkSession) } } - // Projected roots are matched by field ID so historical snapshots still identify renamed - // columns. Both schema checks below use them. - val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol) - val resolver = session.sessionState.conf.resolver - val projectedFieldIds = projectedDataColumns.map { attr => - metadata.globalFieldIdMapping.collectFirst { - case (fieldName, fieldId) if resolver(fieldName, attr.name) => fieldId - } - } - val resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet - val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty) - // iceberg-rust applies name mappings only at the root (#6790). Nested columns can // disappear during projection, or reach the adapter with physical field names that // cannot distinguish a mapped alias from a genuinely missing field. Schema history diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 8534b7f1761..b8835212129 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -1224,6 +1224,80 @@ class CometIcebergNativeSuite } } + // iceberg-rust fails to read a nested field that has a default value and that a data file lacks. + // While nested schema pruning is off, the scan reads every nested field of a projected column, + // including one the query prunes away, so the check on defaults must cover the task schema. + test("nested field with a default value falls back when the scan reads it") { + assume(icebergAvailable, "Iceberg not available in classpath") + assume(icebergVersionAtLeast(1, 11), "Nested column defaults require Iceberg 1.11+") + + 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_default" + spark.sql( + s"CREATE TABLE $table (id INT, s STRUCT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '3')") + spark.sql( + s"INSERT INTO $table VALUES (1, named_struct('a', 1)), (2, named_struct('a', 2))") + addIcebergIntColumnWithDefault( + loadIcebergTable(spark, "test_cat", "db", "nested_default"), + "b", + 7, + parent = Some("s")) + spark.sql(s"REFRESH TABLE $table") + + // The query prunes s.b, so the pruned task schema omits it. + checkIcebergNativeScan(s"SELECT id, s.a FROM $table ORDER BY id") + // With pruning off the scan reads s.b too, and the file lacks it. + withSQLConf(CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") { + checkIcebergNativeScanFallback( + s"SELECT id, s.a FROM $table ORDER BY id", + "nested default read by the unpruned task schema") + } + checkIcebergNativeScanFallback( + s"SELECT id, s.b FROM $table ORDER BY id", + "nested default read by the query") + spark.sql(s"DROP TABLE $table") + } + } + } + + // iceberg-rust binds an equality delete's key by name, and rejects a case-insensitive name that + // matches two columns. The delete's key is taken from the task schema, spelled exactly as in it, + // so the binding is case sensitive. The query reads neither column, as the batch adaptation + // after iceberg-rust rejects a read of two columns that differ only by case. + test("equality delete on a column whose name differs from another only by case") { + 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, + SQLConf.CASE_SENSITIVE.key -> "true", + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val table = "test_cat.db.eq_delete_case" + spark.sql( + s"CREATE TABLE $table (id INT, ID STRING, v STRING) USING iceberg " + + "TBLPROPERTIES ('format-version' = '2')") + spark.sql(s"INSERT INTO $table VALUES (1, 'a', 'x'), (2, 'b', 'y'), (3, 'c', 'z')") + commitEqualityDelete("test_cat", "db", "eq_delete_case", "id", 2, warehouseDir) + + // The pruned task schema holds only `v` and the key `id`. The full one holds both columns. + checkPrunedAndFullReads(s"SELECT v FROM $table ORDER BY v") + spark.sql(s"DROP TABLE $table") + } + } + } + // 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") diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala b/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala index 179a53e0915..f85917bb142 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala @@ -124,13 +124,16 @@ trait CometIcebergTestBase { this: CometTestBase => /** * Adds an `int` column whose initial and write defaults are both `defaultValue`, through - * `UpdateSchema.addColumn(name, type, Literal)`. Requires Iceberg 1.10+ and a v3 table. Callers - * must `REFRESH TABLE` afterwards. + * `UpdateSchema.addColumn(name, type, Literal)`, or to the struct `parent` through + * `addColumn(parent, name, type, Literal)`. Requires Iceberg 1.10+ and a v3 table. Callers must + * `REFRESH TABLE` afterwards. */ protected def addIcebergIntColumnWithDefault( icebergTable: AnyRef, columnName: String, - defaultValue: Int): Unit = { + defaultValue: Int, + parent: Option[String] = None): Unit = { + val typeClass = IcebergReflection.loadClass("org.apache.iceberg.types.Type") val intType = IcebergReflection .loadClass("org.apache.iceberg.types.Types$IntegerType") .getMethod("get") @@ -139,18 +142,22 @@ trait CometIcebergTestBase { this: CometTestBase => .loadClass("org.apache.iceberg.expressions.Expressions") .getMethod("lit", classOf[Object]) .invoke(null, Integer.valueOf(defaultValue)) + val literalClass = IcebergReflection.loadClass("org.apache.iceberg.expressions.Literal") val update = IcebergReflection .loadClass("org.apache.iceberg.Table") .getMethod("updateSchema") .invoke(icebergTable) val updateSchemaClass = IcebergReflection.loadClass("org.apache.iceberg.UpdateSchema") - updateSchemaClass - .getMethod( - "addColumn", - classOf[String], - IcebergReflection.loadClass("org.apache.iceberg.types.Type"), - IcebergReflection.loadClass("org.apache.iceberg.expressions.Literal")) - .invoke(update, columnName, intType, literal) + parent match { + case Some(struct) => + updateSchemaClass + .getMethod("addColumn", classOf[String], classOf[String], typeClass, literalClass) + .invoke(update, struct, columnName, intType, literal) + case None => + updateSchemaClass + .getMethod("addColumn", classOf[String], typeClass, literalClass) + .invoke(update, columnName, intType, literal) + } updateSchemaClass.getMethod("commit").invoke(update) }