diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index b24bc258539..c07d8ee0867 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -204,10 +204,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 where no schema in the table's history gives a partition source or equality-delete key a name that the scan's other columns don't use (every task of a native scan must give such a column the same name) @@ -779,14 +779,9 @@ Which entries apply depends on which native paths are enabled: ### Reads that differ from Spark -- **Migrated tables that use a name mapping.** The `migrate`, `snapshot` and `add_files` - procedures set `schema.name-mapping.default` for files without Iceberg field ids. When a query - also reads a top-level primitive column, the native scan returns NULL for nested fields (struct - fields, list elements and map values). Filters on those fields then select the wrong rows - ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The fix needs an iceberg-rust - upgrade ([#6934](https://github.com/apache/datafusion-comet/issues/6934)). - **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 a28ff0aaf1b..e273a15b156 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -3589,7 +3589,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", @@ -3647,7 +3647,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", @@ -3657,7 +3657,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 498e3c7414e..1fb5324f449 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 33a0593568d..e7c094d8b0b 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 bf0ccd615e8..2efa5db6717 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -789,16 +789,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 @@ -808,18 +809,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 85bc0daa1ed..531fec52e1c 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.DynamicPruningExpression @@ -120,13 +120,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") @@ -1801,11 +1812,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 => @@ -1827,34 +1834,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) @@ -1868,12 +1868,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))") @@ -1886,7 +1885,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. @@ -1901,7 +1900,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 => @@ -1919,19 +1918,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") @@ -1939,14 +1931,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") @@ -1954,9 +1942,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") @@ -1965,7 +1953,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 => @@ -1977,9 +1965,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))") @@ -1992,14 +1980,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 => @@ -2020,8 +2008,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)), @@ -2040,15 +2027,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") } @@ -2160,6 +2212,66 @@ class CometIcebergNativeSuite } } + // SparkActions.snapshotTable sets a name mapping, and the files have no field ids, so the native + // read applies the mapping to the nested fields as well as the top-level ones. + test("migration - nested fields are read through the name mapping") { + 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 sourceName = "parquet_nested_source" + val destName = "test_cat.db.iceberg_nested_dest" + val dataPath = s"${warehouseDir.getAbsolutePath}/nested_source_data" + + spark + .sql("""SELECT CAST(id AS INT) AS id, + named_struct('a', CAST(id AS INT), 'b', CAST(-id AS INT)) AS s, + array(named_struct('x', CAST(id AS INT))) AS items, + map('k', named_struct('v', CAST(id AS INT))) AS m + FROM range(50)""") + .coalesce(1) + .write + .mode("overwrite") + .option("path", dataPath) + .saveAsTable(sourceName) + + try { + val actionsClass = Class.forName("org.apache.iceberg.spark.actions.SparkActions") + val actions = actionsClass.getMethod("get").invoke(null) + val snapshotAction = + actions.getClass + .getMethod("snapshotTable", classOf[String]) + .invoke(actions, sourceName) + val snapshotWithDest = + snapshotAction.getClass + .getMethod("as", classOf[String]) + .invoke(snapshotAction, destName) + snapshotWithDest.getClass.getMethod("execute").invoke(snapshotWithDest) + + checkIcebergNativeScan(s"SELECT id, s.a, s.b FROM $destName WHERE id < 3 ORDER BY id") + checkIcebergNativeScan(s"SELECT id, s.b FROM $destName WHERE id < 3 ORDER BY id") + checkIcebergNativeScan(s"SELECT id, s FROM $destName WHERE id < 3 ORDER BY id") + checkIcebergNativeScan( + s"SELECT id, items.x, m['k'].v FROM $destName WHERE id < 3 ORDER BY id") + checkIcebergNativeScan(s"SELECT s.a, s.b FROM $destName ORDER BY s.a LIMIT 3") + + spark.sql(s"DROP TABLE $destName") + spark.sql(s"DROP TABLE $sourceName") + } catch { + case _: ClassNotFoundException => + cancel("Iceberg Actions API not available - requires iceberg-spark-runtime") + } + } + } + } + test("migration - hive-style partitioned table has partition values") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -2743,13 +2855,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")