From baf26a1f4bf9dd16bedbcdeaf0a9adb5ab8eb4a3 Mon Sep 17 00:00:00 2001 From: Yunchi Pang Date: Fri, 30 Oct 2026 21:39:22 -0400 Subject: [PATCH 1/4] test: cover a native Iceberg read of a data file that omits an optional nested field --- .../comet/CometIcebergNativeSuite.scala | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index b8835212129..ddddf690a48 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -2241,6 +2241,47 @@ 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() + } + + 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") From 801755ba14395ac3fa1aa8efbb568e8d93a8347b Mon Sep 17 00:00:00 2001 From: Yunchi Pang Date: Sun, 1 Nov 2026 11:59:52 -0500 Subject: [PATCH 2/4] test: assert the narrow Iceberg data file omits the nested fields --- .../comet/CometIcebergNativeSuite.scala | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index ddddf690a48..9da62610548 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -2274,6 +2274,26 @@ class CometIcebergNativeSuite .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") From df8300b87786c66b7fdf41447cbff7d8fd961c2b Mon Sep 17 00:00:00 2001 From: Yunchi Pang Date: Wed, 4 Nov 2026 13:52:17 -0500 Subject: [PATCH 3/4] docs: remove issue #7033 --- docs/source/user-guide/latest/iceberg.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 13f811a68c6..98ec551c99c 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -829,10 +829,6 @@ The physical differences that every native data file has, and why they are safe, ### Reads that fail where Spark succeeds -- **A data file that omits an optional nested field.** The native scan fails on a data file that - lacks an optional struct field, which a write with `write.spark.accept-any-schema=true` can - produce without any schema change - ([#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)). - **Column names that only newer Unicode case data folds together**, such as a pair of Vithkuqi From edb9ddf17d51eef0867ab37444d38473ffab32ec Mon Sep 17 00:00:00 2001 From: Yunchi Pang Date: Fri, 6 Nov 2026 08:39:00 -0500 Subject: [PATCH 4/4] test: share Parquet footer reader helper in CometIcebergNativeSuite --- .../comet/CometIcebergNativeSuite.scala | 55 ++++++++----------- 1 file changed, 23 insertions(+), 32 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 9da62610548..629f2458842 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -26,10 +26,14 @@ import java.util.UUID import java.util.concurrent.atomic.AtomicLong import scala.jdk.CollectionConverters._ +import scala.util.Using +import org.apache.hadoop.fs.Path import org.apache.iceberg.data.IcebergGenerics import org.apache.iceberg.expressions.Expressions import org.apache.iceberg.spark.Spark3Util +import org.apache.parquet.hadoop.ParquetFileReader +import org.apache.parquet.hadoop.util.HadoopInputFile import org.apache.spark.{CometListenerBusUtils, SparkException} import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{CometTestBase, DataFrame, Row} @@ -162,6 +166,12 @@ class CometIcebergNativeSuite count } + /** Opens the Parquet footer of the file at `path`, runs `f` on it, and always closes it. */ + private def withParquetReader[T](path: String)(f: ParquetFileReader => T): T = + Using.resource( + ParquetFileReader.open( + HadoopInputFile.fromPath(new Path(path), spark.sessionState.newHadoopConf())))(f) + test("create and query simple Iceberg table with Hadoop catalog") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -1021,24 +1031,17 @@ class CometIcebergNativeSuite // Check the fixture shape in the footer so it cannot silently degrade into a layout // where file-level or row-group-level pruning would explain the byte savings. - val reader = org.apache.parquet.hadoop.ParquetFileReader.open( - org.apache.parquet.hadoop.util.HadoopInputFile.fromPath( - new org.apache.hadoop.fs.Path(dataFilePath), - spark.sessionState.newHadoopConf())) - val idPages = - try { - val rowGroups = reader.getRowGroups - assert(rowGroups.size == 1, s"expected one row group, got ${rowGroups.size}") - val idChunk = rowGroups - .get(0) - .getColumns - .asScala - .find(_.getPath.toDotString == "id") - .getOrElse(fail("id column chunk not found")) - reader.readOffsetIndex(idChunk).getPageCount - } finally { - reader.close() - } + val idPages = withParquetReader(dataFilePath) { reader => + val rowGroups = reader.getRowGroups + assert(rowGroups.size == 1, s"expected one row group, got ${rowGroups.size}") + val idChunk = rowGroups + .get(0) + .getColumns + .asScala + .find(_.getPath.toDotString == "id") + .getOrElse(fail("id column chunk not found")) + reader.readOffsetIndex(idChunk).getPageCount + } assert(idPages > 50, s"expected more than 50 id pages, got $idPages") def runAndCollectScan(sql: String): (CometIcebergNativeScanExec, Long) = { @@ -2276,16 +2279,10 @@ class CometIcebergNativeSuite // 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 { + withParquetReader(row.getString(0)) { reader => reader.getFooter.getFileMetaData.getSchema.getColumns.asScala .map(_.getPath.mkString(".")) .toSet - } finally { - reader.close() } } val omitted = Set("s.b", "l.list.element.y") @@ -4650,11 +4647,7 @@ class CometIcebergNativeSuite assert(parquetFiles.nonEmpty, "Expected at least one Parquet file") val parquetFile = parquetFiles.head - val reader = org.apache.parquet.hadoop.ParquetFileReader.open( - org.apache.parquet.hadoop.util.HadoopInputFile.fromPath( - new org.apache.hadoop.fs.Path(parquetFile.getAbsolutePath), - spark.sessionState.newHadoopConf())) - try { + withParquetReader(parquetFile.getAbsolutePath) { reader => val parquetSchema = reader.getFooter.getFileMetaData.getSchema val int96Columns = parquetSchema.getColumns.asScala .filter(_.getPrimitiveType.getPrimitiveTypeName == @@ -4664,8 +4657,6 @@ class CometIcebergNativeSuite assert( int96Columns.size >= 4, s"Expected at least 4 INT96 columns but found ${int96Columns.size}: ${int96Columns.mkString(", ")}") - } finally { - reader.close() } // Create an unpartitioned Iceberg table and import the Parquet files