diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index b8835212129..9da62610548 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -2241,6 +2241,67 @@ class CometIcebergNativeSuite } } + // No schema change, so the schema evolution fallback does not apply. + test("data file that omits an optional nested field stays native") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withHadoopCatalog("test_cat") { + withSQLConf( + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val table = "test_cat.db.omitted_nested" + spark.sql(s""" + CREATE TABLE $table (id INT, s STRUCT, l ARRAY>) + USING iceberg + """) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.sql(s""" + INSERT INTO $table VALUES + (1, named_struct('a', 1, 'b', 10), array(named_struct('x', 1, 'y', 100))) + """) + // Set after the insert: accept-any-schema matches columns by name. + spark.sql( + s"ALTER TABLE $table SET TBLPROPERTIES ('write.spark.accept-any-schema' = 'true')") + // This file has no s.b and no l.element.y. + spark + .sql(""" + SELECT 2 AS id, named_struct('a', 2) AS s, array(named_struct('x', 20)) AS l + UNION ALL + SELECT 3, CAST(NULL AS STRUCT), CAST(NULL AS ARRAY>) + """) + .writeTo(table) + .append() + } + + // Check the footers, so the test fails if a writer starts filling in the missing fields. + val fileLeaves = spark.sql(s"SELECT file_path FROM $table.files").collect().map { row => + val reader = org.apache.parquet.hadoop.ParquetFileReader.open( + org.apache.parquet.hadoop.util.HadoopInputFile.fromPath( + new org.apache.hadoop.fs.Path(row.getString(0)), + spark.sessionState.newHadoopConf())) + try { + reader.getFooter.getFileMetaData.getSchema.getColumns.asScala + .map(_.getPath.mkString(".")) + .toSet + } finally { + reader.close() + } + } + val omitted = Set("s.b", "l.list.element.y") + // Check the paths against the full-schema file first, or a wrong path makes the next + // assert always pass. + assert(fileLeaves.exists(omitted.subsetOf), s"leaves: ${fileLeaves.mkString(", ")}") + assert(fileLeaves.exists(!_.exists(omitted)), s"leaves: ${fileLeaves.mkString(", ")}") + + checkPrunedAndFullReads(s"SELECT * FROM $table ORDER BY id") + checkPrunedAndFullReads(s"SELECT id, s.b FROM $table WHERE s.b IS NULL ORDER BY id") + + spark.sql(s"DROP TABLE $table") + } + } + } + test("schema evolution - nested field dropped, moved, or promoted stays native") { assume(icebergAvailable, "Iceberg not available in classpath")