You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/16 14:30:45 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6725-283b185b69573ed5f46c0523dc5e03831d6cd4b1 created (now ad327b6d60)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6725-283b185b69573ed5f46c0523dc5e03831d6cd4b1
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at ad327b6d60 perf: read only the nested fields Spark keeps in the native Iceberg scan (#6725)
This branch includes the following new commits:
new ad327b6d60 perf: read only the nested fields Spark keeps in the native Iceberg scan (#6725)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: perf: read only the nested fields Spark keeps in the native Iceberg scan (#6725)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6725-283b185b69573ed5f46c0523dc5e03831d6cd4b1
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit ad327b6d60e582afe1fb0263597b676fd4b9235c
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Fri Oct 16 14:30:16 2026 +0000
perf: read only the nested fields Spark keeps in the native Iceberg scan (#6725)
* perf: read only the nested fields Spark keeps in the native Iceberg scan
The native Iceberg scan sent iceberg-rust the full table schema as the
FileScanTask schema, so iceberg-rust read and decoded every leaf of a
projected struct, list, or map column even when Spark's nested schema
pruning kept only some of them. Pass the pruned scan schema instead,
without metadata columns, and still add partition sources and
equality-delete keys. Memoize those additions so tasks share one schema
in the pool, and keep the full schema when a needed column is nested
inside a pruned struct.
Tasks with deletes read with the pruned schema too, so a VERSION AS OF
read of a column dropped after the snapshot no longer fails natively
when the table has deletes. The pruned schema is built at planning time,
where a reflection failure falls back to Spark, and findFieldObject now
propagates reflection failures. A column added to a task schema resolves
from the current table schema before the schema history, so it keeps its
current name and type.
Add spark.comet.scan.icebergNative.nestedSchemaPruning.enabled to turn
it off.
* feedback
* Resolve appended task-schema columns from one schema
Take every missing partition source and equality-delete key from one
schema, the current one or the newest in the history that has them all
under names the task schema does not use. Picking a name per field
clashed when two partition sources were renamed into each other's names.
Tasks that need no extra columns skip the memo lookups, and tasks
without deletes skip the equality-key work.
* Build one task schema per scan for native Iceberg reads
iceberg-rust caches an equality-delete predicate per delete file, with
the column names of the task that loads it first, and binds it by name
in every task of a reader. Task schemas built per task could name a key
differently across tasks, which failed the scan or, after two columns
swapped names, deleted the wrong rows.
Build one pruned and one full task schema per scan during planning. The
names come from the newest schema whose top-level columns match the scan
schema, and an equality key dropped before it takes one name from the
schema history. A scan whose keys cannot be named that way falls back to
Spark. Serialization now builds the task schema and the projection once
per scan.
* Check the full task schema for nested evolution
The full task schema of a VERSION AS OF read can be older than the
table's schema and hold a nested field that an earlier data file lacks,
which the native scan cannot read. Walk it in the nested-evolution check
too, so such a scan falls back to Spark.
* Check the task schema the scan reads for nested evolution
The nested-evolution check now walks the schema the scan reads, the
pruned task schema while pruning applies and the full one otherwise,
through one method that serialization uses too. A pruned VERSION AS OF
read no longer falls back because of a nested field only the full
schema holds.
The Iceberg user guide lists the new naming fallback. The SQL fixture
adds empty lists and maps, and the pruned-read tests also check the
answers with pruning off.
* Match field types when picking the full task schema
The full task schema was the newest schema whose top-level columns
match the scan schema by id and name. For a VERSION AS OF read across a
type promotion, that could be a newer schema with the promoted type,
which the batch adaptation after iceberg-rust cannot narrow back, so
the native scan failed. Pick the newest schema in which every field the
scan reads, at any depth, has the same name and type instead.
The time travel test reads a promoted int, decimal, and nested field,
on partitioned and unpartitioned tables, with pruning on and off.
* Fix the native Iceberg task schema at planning
`taskSchema` read the nested schema pruning config every time it was
called: during planning, where CometScanRule checks the schema, and
again when the scan ran and serialized it. A query planned with pruning
on and run with pruning off, such as a cached DataFrame, read the full
schema, which the nested-evolution check never walked. Make it a val,
built with the metadata during planning.
The struct fallback test plans a VERSION AS OF read with pruning on and
runs it with pruning off.
* Simplify the native Iceberg task schema code and tests
Build the scan schema's data columns and the projected field ids in
one pass each, and name the type check's schema `fullTaskSchema` so it
reads apart from `taskSchema`, the schema the scan reads. The promotion
reads reuse `checkPrunedAndFullReads`, and a time travel query that
could not catch a wrong schema choice is gone.
---
docs/source/user-guide/latest/iceberg.md | 3 +
.../main/scala/org/apache/comet/CometConf.scala | 10 +
.../apache/comet/iceberg/IcebergReflection.scala | 244 +++++++---
.../org/apache/comet/rules/CometScanRule.scala | 78 +++-
.../serde/operator/CometIcebergNativeScan.scala | 129 ++----
.../serde/operator/CometIcebergNativeWrite.scala | 4 +-
.../sql-tests/iceberg/nested_schema_pruning.sql | 136 ++++++
.../org/apache/comet/CometIcebergNativeSuite.scala | 490 ++++++++++++++++++++-
.../sql/benchmark/CometIcebergReadBenchmark.scala | 109 ++++-
9 files changed, 1006 insertions(+), 197 deletions(-)
diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index 24713a32b7..0febcbb918 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -197,6 +197,9 @@ The following scenarios will fall back to the JVM Iceberg reader:
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 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)
Writes are not covered by this list. By default Spark plans Iceberg writes with its own V2 write
operator, and iceberg-java writes the data files. With `spark.comet.write.iceberg.enabled=true`,
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index 2b48d26b40..4ee51c5c08 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -155,6 +155,16 @@ object CometConf extends ShimCometConf {
.checkValue(v => v > 0, "Data file concurrency limit must be positive")
.createWithDefault(1)
+ val COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.scan.icebergNative.nestedSchemaPruning.enabled")
+ .category(CATEGORY_SCAN)
+ .doc(
+ "Whether the native Iceberg scan reads only the nested fields of a struct, list, or map " +
+ "column that Spark's nested schema pruning keeps. When false, it reads every nested " +
+ "field of a projected column and drops the unused ones after decoding.")
+ .booleanConf
+ .createWithDefault(true)
+
val COMET_ICEBERG_IO_TIMEOUT: ConfigEntry[Long] =
conf("spark.comet.iceberg.ioTimeout")
.category(CATEGORY_SCAN)
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 354ff37e21..89ef99857e 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -28,6 +28,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
+import org.apache.comet.CometConf
import org.apache.comet.util.ClassLoaders
/**
@@ -707,89 +708,163 @@ object IcebergReflection extends Logging {
}
}
- /** Returns the `Types.NestedField` for `fieldId` in `schema`, or None. */
- def findFieldObject(schema: Any, fieldId: Int): Option[Any] = {
- try {
- val findFieldMethod = getMethod(schema.getClass, "findField", classOf[Int])
- Option(findFieldMethod.invoke(schema, fieldId.asInstanceOf[AnyRef]))
- } catch {
- case _: Exception => None
+ /**
+ * Returns the `Types.NestedField` for `fieldId` at any depth of `schema`, or None when `schema`
+ * has no such field. A reflection failure propagates rather than reading as a missing field.
+ */
+ def findFieldObject(schema: Any, fieldId: Int): Option[Any] =
+ Option(
+ getMethod(schema.getClass, "findField", classOf[Int])
+ .invoke(schema, fieldId.asInstanceOf[AnyRef]))
+
+ /**
+ * The task schemas of a native Iceberg scan, built once for all of its tasks during planning:
+ * the pruned scan schema, and the full schema read when pruning is off or cannot apply. Both
+ * hold the partition sources and equality-delete keys (`requiredFieldIds`) the tasks need
+ * beyond the projection: iceberg-rust resolves a task's partition spec against its schema, and
+ * reads the keys to apply equality deletes. The pruned one is None when a required column is
+ * nested inside a struct the query pruned away, as appending it at the top level would put it
+ * outside its struct. The full one is None, so that the scan falls back, when no name fits.
+ *
+ * Every task reads with the same schema because iceberg-rust builds an equality-delete
+ * predicate once per delete file, with the column names of the task that loads it first, and
+ * binds it by name in every other task of the reader. The full schema is the newest schema,
+ * current first, that has every field the scan reads, at any depth, with the same name and
+ * type, as the batch adaptation after iceberg-rust cannot narrow a promoted type back. That is
+ * the current schema unless a VERSION AS OF snapshot named or typed a field differently. A key
+ * that schema lacks, dropped before it, takes the newest name that no other column has.
+ */
+ def nativeTaskSchemas(
+ table: Any,
+ tableSchema: Any,
+ scanSchema: Any,
+ requiredFieldIds: Set[Int]): (Option[Any], Option[Any]) = {
+ val prunedScanSchema = withoutMetadataColumns(scanSchema)
+ val scanFields = fieldsAtAnyDepth(prunedScanSchema)
+ val schemas = tableSchema +: getAllSchemas(table).reverse
+ val full = schemas.find { schema =>
+ scanFields.forall { case (id, name, leafType) =>
+ findFieldObject(schema, id).exists { field =>
+ fieldName(field) == name && leafType.forall(_ == fieldTypeOf(field))
+ }
+ }
}
+ full match {
+ case None => (None, None)
+ case Some(fullSchema) =>
+ val fullColumns = columnNamesById(fullSchema)
+ val fullNames = fullColumns.values.toSet
+ val ids = requiredFieldIds.toSeq.sorted
+ val inFull = ids.flatMap(id => findFieldObject(fullSchema, id).map(id -> _)).toMap
+ // Keys dropped before `fullSchema`, each under a name that no other column has.
+ val dropped =
+ ids.filterNot(inFull.contains).foldLeft(Option(Map.empty[Int, Any])) { (resolved, id) =>
+ resolved.flatMap { fields =>
+ val taken = fullNames ++ fields.values.map(fieldName)
+ schemas.iterator
+ .flatMap(findFieldObject(_, id))
+ .find(field => !taken.contains(fieldName(field)))
+ .map(field => fields + (id -> field))
+ }
+ }
+ dropped match {
+ case None => (None, None)
+ case Some(droppedFields) =>
+ val missing = ids.filter(findFieldObject(prunedScanSchema, _).isEmpty)
+ val prunedTaskSchema =
+ if (missing.forall(id => fullColumns.contains(id) || !inFull.contains(id))) {
+ val fields = missing.map(id => inFull.getOrElse(id, droppedFields(id)))
+ Some(withColumns(prunedScanSchema, fields))
+ } else {
+ None
+ }
+ val fullFields = droppedFields.toSeq.sortBy(_._1).map(_._2)
+ (prunedTaskSchema, Some(withColumns(fullSchema, fullFields)))
+ }
+ }
+ }
+
+ /** The top-level columns of `schema`, as field id to name. */
+ private def columnNamesById(schema: Any): Map[Int, String] = {
+ import scala.jdk.CollectionConverters._
+ getMethod(schema.getClass, "columns")
+ .invoke(schema)
+ .asInstanceOf[java.util.List[_]]
+ .asScala
+ .map(column => fieldIdOf(column) -> fieldName(column))
+ .toMap
}
/**
- * Returns a schema equal to `baseSchema` but guaranteed to contain `requiredFieldIds`. Any id
- * not already present is resolved from the table's schema history (`table.schemas()`) and
- * appended.
- *
- * Two callers need this: an equality delete may be keyed on a column since dropped from the
- * current schema (mirroring Iceberg-Java's `DeleteFilter.fileProjection`), and a partition
- * spec's source columns must be present for iceberg-rust to resolve `partition_type` when the
- * query projects them out. Called at serialization time, so it throws on failure (a required id
- * that cannot be resolved, or any reflection error) rather than silently degrading.
- * CometScanRule is responsible for falling back before we get here.
- */
- def schemaWithRequiredFields(baseSchema: Any, table: Any, requiredFieldIds: Seq[Int]): Any = {
- // `findFieldObject` searches recursively, so a source column already present as a nested field
- // (e.g. `s.region`) is not re-appended at the top level, which would create a duplicate field
- // id. Empty `requiredFieldIds` (the common non-partitioned, no-delete task) does no lookups.
- val missingIds =
- requiredFieldIds.distinct.filter(id => findFieldObject(baseSchema, id).isEmpty)
- if (missingIds.isEmpty) {
- baseSchema
+ * The fields of `schema` at any depth, as id, name, and the type of a field with no children.
+ */
+ private def fieldsAtAnyDepth(schema: Any): Seq[(Int, String, Option[Any])] = {
+ import scala.jdk.CollectionConverters._
+ def walk(field: Any): Seq[(Int, String, Option[Any])] = {
+ val children = childFields(field)
+ val leafType = if (children.isEmpty) Some(fieldTypeOf(field)) else None
+ (fieldIdOf(field), fieldName(field), leafType) +: children.flatMap(walk)
+ }
+ getMethod(schema.getClass, "columns")
+ .invoke(schema)
+ .asInstanceOf[java.util.List[_]]
+ .asScala
+ .toSeq
+ .flatMap(walk)
+ }
+
+ /** `schema` with `fields` appended at its top level, or `schema` itself when there are none. */
+ private def withColumns(schema: Any, fields: Seq[Any]): Any =
+ if (fields.isEmpty) {
+ schema
} else {
- logDebug(
- s"Native Iceberg scan schema is missing field id(s) ${missingIds.mkString(",")}; " +
- "resolving them from table schema history")
- val history = getAllSchemas(table)
- val resolvedFields = missingIds.map { id =>
- history.iterator
- .flatMap(s => findFieldObject(s, id))
- .toSeq
- .headOption
- .getOrElse(throw new IllegalStateException(
- s"Cannot resolve field id $id in table schema history"))
- }
- val existing =
- getMethod(baseSchema.getClass, "columns")
- .invoke(baseSchema)
- .asInstanceOf[java.util.List[_]]
- val newColumns = new java.util.ArrayList[Any](existing)
- resolvedFields.foreach(newColumns.add)
- baseSchema.getClass
+ val columns =
+ getMethod(schema.getClass, "columns").invoke(schema).asInstanceOf[java.util.List[_]]
+ val newColumns = new java.util.ArrayList[Any](columns.size + fields.length)
+ newColumns.addAll(columns)
+ fields.foreach(newColumns.add)
+ schema.getClass
.getConstructor(classOf[java.util.List[_]])
.newInstance(newColumns)
.asInstanceOf[AnyRef]
}
- }
+
+ /** The spec reserves field ids above `Integer.MAX_VALUE - 200` for metadata columns. */
+ val MaxDataFieldId: Int = Int.MaxValue - 200
/**
- * The source column field ids referenced by a task's partition spec.
- *
- * iceberg-rust validates a `FileScanTask` by resolving its partition spec against the task
- * schema (`partition_type(schema)`), so a task carrying a partition spec needs those source
- * columns present in the schema even when the query projects them out (for example selecting
- * only `_spec_id` or `_partition`). These ids are unioned into the native task schema via
- * [[schemaWithRequiredFields]]. project_field_ids still drives the read, so the columns are not
- * materialized into the scan output. All ids resolve from the table's schema history, including
- * a source column later dropped by partition evolution.
- *
- * Returns an empty sequence when the task has no partition spec.
+ * Returns `schema` without its top-level metadata columns (`_file`, `_pos`, `_partition` and
+ * the other reserved ids), or `schema` itself when it has none. A scan's expected schema
+ * includes the metadata columns the query selects, which a table schema never has.
*/
- def partitionSourceFieldIds(task: Any, fileScanTaskClass: Class[_]): Seq[Int] = {
- val spec =
- try {
- getMethod(fileScanTaskClass, "spec").invoke(task)
- } catch {
- case _: Exception => null
- }
- if (spec == null) Seq.empty else partitionFieldSourceIds(spec)
+ def withoutMetadataColumns(schema: Any): Any = {
+ import scala.jdk.CollectionConverters._
+ val columns =
+ getMethod(schema.getClass, "columns").invoke(schema).asInstanceOf[java.util.List[_]]
+ val dataColumns = new java.util.ArrayList[Any](columns.size)
+ columns.asScala.iterator
+ .filter(fieldIdOf(_) <= MaxDataFieldId)
+ .foreach(column => dataColumns.add(column))
+ if (dataColumns.size == columns.size) {
+ schema
+ } else {
+ schema.getClass
+ .getConstructor(classOf[java.util.List[_]])
+ .newInstance(dataColumns)
+ .asInstanceOf[AnyRef]
+ }
}
/**
* The `sourceId`s of a partition spec's fields, in spec order. Empty for an unpartitioned spec.
+ *
+ * iceberg-rust validates a `FileScanTask` by resolving its partition spec against the task
+ * schema (`partition_type(schema)`), so the source columns must be in the task schema even when
+ * the query projects them out (for example selecting only `_spec_id` or `_partition`). See
+ * [[nativeTaskSchemas]]. project_field_ids still drives the read, so the columns are not
+ * materialized into the scan output.
*/
- private def partitionFieldSourceIds(spec: Any): Seq[Int] = {
+ def partitionFieldSourceIds(spec: Any): Seq[Int] = {
import scala.jdk.CollectionConverters._
// Exclude fields whose source column was dropped (unknown result type): their source is absent
// from the current schema and the serialized spec/values omit them too (see
@@ -1336,6 +1411,8 @@ object IcebergReflection extends Logging {
private def fieldIdOf(field: Any): Int =
getMethod(field.getClass, "fieldId").invoke(field).asInstanceOf[Int]
+ private def fieldTypeOf(field: Any): Any = getMethod(field.getClass, "type").invoke(field)
+
/**
* Converts an Iceberg `Schema` to the Spark `StructType` it reads as, via
* `SparkSchemaUtil.convert`. Comet serializes the whole table/scan schema to native (not just
@@ -2393,6 +2470,14 @@ object IcebergReflection extends Logging {
* List of FileScanTask objects from Iceberg planning
* @param scanSchema
* The expectedSchema from the SparkScan (for schema evolution / VERSION AS OF)
+ * @param prunedTaskSchema
+ * The schema every task reads with when nested schema pruning is on: scanSchema without its
+ * metadata columns, which iceberg-rust resolves by reserved field id, plus the partition
+ * sources and equality-delete keys the tasks need. None when one of those is nested inside a
+ * struct the query pruned away. See [[IcebergReflection.nativeTaskSchemas]].
+ * @param fullTaskSchema
+ * The schema every task reads with otherwise: the scan's full table schema plus the
+ * equality-delete keys it lacks. None when no name fits every task, which falls back to Spark.
* @param globalFieldIdMapping
* Mapping from column names to Iceberg field IDs (built from scanSchema)
* @param catalogProperties
@@ -2407,11 +2492,26 @@ case class CometIcebergNativeScanMetadata(
nameMapping: Option[String],
@transient tasks: java.util.List[_],
scanSchema: Any,
+ prunedTaskSchema: Option[Any],
+ fullTaskSchema: Option[Any],
tableSchema: Any,
globalFieldIdMapping: Map[String, Int],
catalogProperties: Map[String, String],
catalogName: Option[String],
- fileFormat: String)
+ fileFormat: String) {
+
+ /**
+ * The schema every task of the scan reads with: the pruned task schema while nested schema
+ * pruning is on and can apply, the full one otherwise. It is fixed when `extract` builds the
+ * metadata during planning, so the schema CometScanRule checks is the one
+ * CometIcebergNativeScan serializes at execution, even if the config changes in between. None
+ * falls back to Spark.
+ */
+ val taskSchema: Option[Any] =
+ prunedTaskSchema
+ .filter(_ => CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.get())
+ .orElse(fullTaskSchema)
+}
object CometIcebergNativeScanMetadata extends Logging {
@@ -2431,6 +2531,9 @@ object CometIcebergNativeScanMetadata extends Logging {
* The scan's FileScanTasks (already extracted). Passed in rather than re-read via
* [[IcebergReflection.getTasks]], which for a staged scan rebuilds a flattened list of every
* task on each call.
+ * @param requiredFieldIds
+ * The partition sources and equality-delete keys of those tasks, which every task schema
+ * holds
* @return
* Some(metadata) if all reflection succeeds, None to trigger fallback
*/
@@ -2438,7 +2541,8 @@ object CometIcebergNativeScanMetadata extends Logging {
scan: Any,
metadataLocation: String,
catalogProperties: Map[String, String],
- tasks: java.util.List[_]): Option[CometIcebergNativeScanMetadata] = {
+ tasks: java.util.List[_],
+ requiredFieldIds: Set[Int]): Option[CometIcebergNativeScanMetadata] = {
import org.apache.comet.iceberg.IcebergReflection._
for {
@@ -2457,6 +2561,8 @@ object CometIcebergNativeScanMetadata extends Logging {
}
val globalFieldIdMapping = buildFieldIdMapping(scanSchema)
+ val (prunedTaskSchema, fullTaskSchema) =
+ nativeTaskSchemas(table, tableSchema, scanSchema, requiredFieldIds)
// File format is always PARQUET,
// validated in CometScanRule.validateIcebergFileScanTasks()
@@ -2467,6 +2573,8 @@ object CometIcebergNativeScanMetadata extends Logging {
nameMapping = nameMapping,
tasks = tasks,
scanSchema = scanSchema,
+ prunedTaskSchema = prunedTaskSchema,
+ fullTaskSchema = fullTaskSchema,
tableSchema = tableSchema,
globalFieldIdMapping = globalFieldIdMapping,
catalogProperties = catalogProperties,
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 e65b73d130..5460a1eeb8 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -633,8 +633,12 @@ case class CometScanRule(session: SparkSession)
.get(COMET_S3_COMPLIANT_SCHEMES_KEY)
.map(COMET_S3_COMPLIANT_SCHEMES_KEY -> _)
- val result = CometIcebergNativeScanMetadata
- .extract(scanExec.scan, effectiveLocation, catalogProperties, icebergTasks)
+ val result = CometIcebergNativeScanMetadata.extract(
+ scanExec.scan,
+ effectiveLocation,
+ catalogProperties,
+ icebergTasks,
+ taskValidation.requiredFieldIds)
result
} catch {
@@ -750,15 +754,16 @@ case class CometScanRule(session: SparkSession)
val resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet
val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty)
- // The whole Iceberg table schema is serialized to native, but iceberg-rust can represent
+ // The full task schema can be serialized to native, but iceberg-rust can represent
// Variant in that schema as long as no projected field contains one. Check projected
// roots strictly, and allow Variant only under entirely unprojected roots. Other
// unsupported types still fail closed everywhere. An empty data projection is also strict
// because iceberg-rust currently interprets an empty field-id list as a request for every
// column.
+ val fullTaskSchema = metadata.fullTaskSchema.getOrElse(metadata.tableSchema)
val schemaTypesSupported =
try {
- val fullSchema = IcebergReflection.toSparkSchema(metadata.tableSchema)
+ val fullSchema = IcebergReflection.toSparkSchema(fullTaskSchema)
// DataTypeSupport recursively dispatches back to this override for struct fields,
// array elements, and map entries, so Variant is allowed at any nesting depth only
// when its entire top-level Iceberg field is unprojected.
@@ -769,7 +774,7 @@ case class CometScanRule(session: SparkSession)
reasons: ListBuffer[String]): Boolean =
isVariantType(dt) || super.isTypeSupported(dt, name, reasons)
}
- val tableFieldIds = IcebergReflection.buildFieldIdMapping(metadata.tableSchema)
+ val tableFieldIds = IcebergReflection.buildFieldIdMapping(fullTaskSchema)
fullSchema.fields.forall { field =>
val isProjected = projectedDataColumns.isEmpty ||
@@ -789,15 +794,15 @@ case class CometScanRule(session: SparkSession)
// 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. The native read asks for a
- // projected column's full nested type even when Spark prunes it, taken from the current
- // table schema, or from the scan schema when VERSION AS OF reads a dropped column (see
- // CometIcebergNativeScan). 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.
+ // 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.
val nestedFieldsSupported =
try {
- val schemas = Seq(metadata.tableSchema, metadata.scanSchema)
+ val schemas = Seq(metadata.tableSchema, metadata.scanSchema) ++ metadata.taskSchema
val fieldIds =
if (hasUnresolvedProjectedFieldIds) {
schemas.flatMap(IcebergReflection.buildFieldIdMapping(_).values).toSet
@@ -1054,12 +1059,21 @@ case class CometScanRule(session: SparkSession)
}
}
+ // Every task needs each partition source and equality-delete key under one name (see
+ // IcebergReflection.nativeTaskSchemas).
+ val taskSchemaNamesSupported = metadata.fullTaskSchema.isDefined
+ if (!taskSchemaNamesSupported) {
+ fallbackReasons += "Comet's native Iceberg reader cannot name a partition source or " +
+ "equality-delete key the same way in every task: no schema in the table's history " +
+ "names it apart from the columns the scan reads"
+ }
+
if (schemaSupported && fileIOCompatible && formatVersionSupported &&
defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported &&
taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable &&
metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported &&
transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported &&
- nestedFieldsSupported) {
+ nestedFieldsSupported && taskSchemaNamesSupported) {
CometBatchScanExec(
scanExec.clone().asInstanceOf[BatchScanExec],
runtimeFilters = scanExec.runtimeFilters,
@@ -1372,6 +1386,8 @@ object CometScanRule extends Logging {
IcebergReflection.loadClass(IcebergReflection.ClassNames.FILE_SCAN_TASK)
val unboundPredicateClass =
IcebergReflection.loadClass(IcebergReflection.ClassNames.UNBOUND_PREDICATE)
+ val deleteFileClass =
+ IcebergReflection.loadClass(IcebergReflection.ClassNames.DELETE_FILE)
// Resolve all method lookups outside the loop
val fileMethod = IcebergReflection.getMethod(contentScanTaskClass, "file")
@@ -1379,11 +1395,18 @@ object CometScanRule extends Logging {
val pathMethod = IcebergReflection.getMethod(contentFileClass, "path")
val residualMethod = IcebergReflection.getMethod(contentScanTaskClass, "residual")
val deletesMethod = IcebergReflection.getMethod(fileScanTaskClass, "deletes")
+ val specMethod = IcebergReflection.getMethod(fileScanTaskClass, "spec")
val termMethod = IcebergReflection.getMethod(unboundPredicateClass, "term")
var allParquet = true
val unsupportedSchemes = mutable.Set[String]()
var nonIdentityTransform: Option[String] = None
+ // Many tasks share a partition spec, and a delete file can apply to many data files, so each
+ // distinct instance is inspected once.
+ def identitySet(): java.util.Set[Any] =
+ java.util.Collections.newSetFromMap(new java.util.IdentityHashMap[Any, java.lang.Boolean]())
+ val specs = identitySet()
+ val seenDeleteFiles = identitySet()
val deleteFiles = new java.util.ArrayList[Any]()
// Buckets the native FileIO must read (data + delete files), for the single-config check in
// CometScanRule; only S3-family locations contribute (see NativeConfig.bucketForUri).
@@ -1444,13 +1467,17 @@ object CometScanRule extends Logging {
// Collect delete files and check their schemes
if (fileScanTaskClass.isInstance(task)) {
+ // Unlike the checks around it, a failure here propagates: the spec decides which columns
+ // every task schema needs, and a schema that misses one fails the native scan.
+ Option(specMethod.invoke(task)).foreach(specs.add)
try {
val deletes = deletesMethod.invoke(task).asInstanceOf[java.util.List[_]]
- deleteFiles.addAll(deletes)
-
deletes.asScala.foreach { deleteFile =>
- IcebergReflection.extractFileLocation(contentFileClass, deleteFile).foreach {
- deletePath => inspectLocation(deletePath)
+ if (seenDeleteFiles.add(deleteFile)) {
+ deleteFiles.add(deleteFile)
+ IcebergReflection.extractFileLocation(contentFileClass, deleteFile).foreach {
+ deletePath => inspectLocation(deletePath)
+ }
}
}
} catch {
@@ -1459,13 +1486,25 @@ object CometScanRule extends Logging {
}
}
+ // The columns every native task schema holds beyond the projection: partition sources and
+ // equality-delete keys (see IcebergReflection.nativeTaskSchemas).
+ val requiredFieldIds =
+ specs.asScala.flatMap(IcebergReflection.partitionFieldSourceIds).toSet ++
+ deleteFiles.asScala.flatMap { deleteFile =>
+ IcebergReflection
+ .getEqualityFieldIds(deleteFileClass, deleteFile)
+ .asScala
+ .map(_.asInstanceOf[java.lang.Integer].intValue())
+ }
+
IcebergTaskValidationResult(
allParquet,
unsupportedSchemes.toSet,
nonIdentityTransform,
deleteFiles,
dataFileBuckets.toSet,
- hostlessLocation)
+ hostlessLocation,
+ requiredFieldIds)
}
}
@@ -1478,4 +1517,5 @@ case class IcebergTaskValidationResult(
nonIdentityTransform: Option[String],
deleteFiles: java.util.List[_],
dataFileBuckets: Set[String],
- hostlessLocation: Option[String])
+ hostlessLocation: Option[String],
+ requiredFieldIds: Set[Int])
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index 22895e5145..4070d78536 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -980,10 +980,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
val commonBuilder = OperatorOuterClass.IcebergScanCommon.newBuilder()
// Deduplication structures - map unique values to pool indices
- val schemaToPoolIndex = mutable.HashMap[AnyRef, Int]()
val partitionSpecToPoolIndex = mutable.HashMap[String, Int]()
val nameMappingToPoolIndex = mutable.HashMap[String, Int]()
- val projectFieldIdsToPoolIndex = mutable.HashMap[Seq[Int], Int]()
val partitionDataToPoolIndex = mutable.HashMap[String, Int]()
// Individual delete files are interned into a flat pool; deleteFilesToPoolIndex then dedups
// the per-task sets as lists of indices into it, so a delete file that applies to many data
@@ -1006,26 +1004,39 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
idx
})
val residualToPoolIndex = mutable.HashMap[OperatorOuterClass.IcebergPredicate, Int]()
- // Field-id mappings are read out of an Iceberg schema by reflection, one lookup per column, so
- // memoize them. Keyed like schemaToPoolIndex above: a task schema that Iceberg materializes
- // fresh per task misses, but the table/scan schema shared by every task hits.
- val fieldIdMappingCache = mutable.HashMap[AnyRef, Map[String, Int]]()
- def fieldIdMapping(schema: AnyRef): Map[String, Int] =
- fieldIdMappingCache.getOrElseUpdate(schema, IcebergReflection.buildFieldIdMapping(schema))
- // Whether the scan schema references field ids the current table schema no longer has (a
- // dropped column read through VERSION AS OF). Loop-invariant, and lazy so a scan whose tasks
- // all carry deletes never walks the table schema at all.
- lazy val hasHistoricalColumns = {
- val tableSchemaFieldIds =
- fieldIdMapping(metadata.tableSchema.asInstanceOf[AnyRef]).values.toSet
- metadata.globalFieldIdMapping.values.exists(id => !tableSchemaFieldIds.contains(id))
- }
// Columns whose Iceberg type iceberg-rust cannot use for page-index pruning; residual
// predicates over them are dropped (see icebergExprToProto). Computed once from the full table
// schema so a filter column projected out of the scan output is still recognized.
val pageIndexUnsupportedColumns =
IcebergReflection.pageIndexUnsupportedColumns(metadata.tableSchema)
+ // iceberg-rust reads every leaf of a projected column that the task schema holds. Reading with
+ // the pruned task schema skips the nested fields Spark's schema pruning removed, which the
+ // full schema would read and decode only to drop later. Every task reads with the same schema
+ // (see IcebergReflection.nativeTaskSchemas), so its pool entry and the projected field ids are
+ // the same for every task too.
+ val taskSchema: AnyRef = metadata.taskSchema
+ .getOrElse(
+ throw new IllegalStateException(
+ "No native Iceberg task schema. CometScanRule should have fallen back to Spark."))
+ .asInstanceOf[AnyRef]
+ val nameToFieldId = IcebergReflection.buildFieldIdMapping(taskSchema)
+ val projectFieldIds = OperatorOuterClass.ProjectFieldIdList.newBuilder()
+ output.foreach { attr =>
+ projectFieldIds.addFieldIds(
+ nameToFieldId
+ .get(attr.name)
+ .orElse(metadata.globalFieldIdMapping.get(attr.name))
+ .orElse(CometIcebergNativeScan.MetadataFieldIds.get(attr.name))
+ .getOrElse {
+ throw new IllegalStateException(
+ s"Column '${attr.name}' not found in task schema, global schema, " +
+ "or metadata field IDs. This indicates a bug in CometScanRule " +
+ "validation -- all output columns should be resolvable.")
+ })
+ }
+ commonBuilder.addProjectFieldIdsPool(projectFieldIds.build())
+
val perPartitionBuilders = mutable.ArrayBuffer[OperatorOuterClass.IcebergScan]()
var totalTasks = 0
@@ -1069,9 +1080,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
// keyMetadata() is declared on ContentFile (present across all supported Iceberg versions).
// For encrypted tables it returns the plaintext StandardKeyMetadata blob; null otherwise.
val keyMetadataMethod = IcebergReflection.getMethod(contentFileClass, "keyMetadata")
- val taskSchemaMethod = IcebergReflection.getMethod(fileScanTaskClass, "schema")
val toJsonMethod =
IcebergReflection.getMethod(schemaParserClass, "toJson", schemaClass)
+ commonBuilder.addSchemaPool(toJsonMethod.invoke(null, taskSchema).asInstanceOf[String])
// Access inputRDD - safe now, DPP is resolved
scanExec.inputRDD match {
@@ -1130,83 +1141,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
// verbatim for iceberg-rust to decode. Unencrypted files leave the field unset.
keyMetadataBytes(keyMetadataMethod, dataFile).foreach(taskBuilder.setKeyMetadata)
- val taskSchema = taskSchemaMethod.invoke(task)
-
- val deletes =
- IcebergReflection.getDeleteFilesFromTask(task, fileScanTaskClass)
- val hasDeletes = !deletes.isEmpty
-
- val baseSchema: AnyRef =
- if (hasDeletes) {
- // An equality delete may be keyed on a column dropped from the current schema
- // (schema evolution). iceberg-rust must read that column to apply the delete,
- // so union the equality-delete field ids into the task schema, resolving any
- // dropped ones from the table's schema history (mirrors Iceberg-Java's
- // DeleteFilter.fileProjection).
- val equalityFieldIds = deletes.asScala.flatMap { df =>
- requiredEqualityFieldIds(deleteFileClass, df).asScala.map(_.intValue())
- }.toSeq
- if (equalityFieldIds.nonEmpty) {
- IcebergReflection
- .schemaWithRequiredFields(taskSchema, metadata.table, equalityFieldIds)
- .asInstanceOf[AnyRef]
- } else {
- taskSchema
- }
- } else {
- if (hasHistoricalColumns) {
- metadata.scanSchema.asInstanceOf[AnyRef]
- } else {
- metadata.tableSchema.asInstanceOf[AnyRef]
- }
- }
-
- // iceberg-rust validates a FileScanTask by resolving its partition spec against
- // the task schema, so a task carrying a partition spec needs that spec's source
- // columns present even when the query projects them out (e.g. selecting only
- // _spec_id / _partition). Union them in. project_field_ids still drives the read,
- // so these columns are not materialized into the output.
- val schema: AnyRef =
- IcebergReflection
- .schemaWithRequiredFields(
- baseSchema,
- metadata.table,
- IcebergReflection.partitionSourceFieldIds(task, fileScanTaskClass))
- .asInstanceOf[AnyRef]
-
- val schemaIdx = schemaToPoolIndex.getOrElseUpdate(
- schema, {
- val idx = schemaToPoolIndex.size
- val schemaJson = toJsonMethod.invoke(null, schema).asInstanceOf[String]
- commonBuilder.addSchemaPool(schemaJson)
- idx
- })
- taskBuilder.setSchemaIdx(schemaIdx)
-
- val nameToFieldId = fieldIdMapping(schema)
-
- val projectFieldIds = output.map { attr =>
- nameToFieldId
- .get(attr.name)
- .orElse(metadata.globalFieldIdMapping.get(attr.name))
- .orElse(CometIcebergNativeScan.MetadataFieldIds.get(attr.name))
- .getOrElse {
- throw new IllegalStateException(
- s"Column '${attr.name}' not found in task schema, global schema, " +
- "or metadata field IDs. This indicates a bug in CometScanRule " +
- "validation -- all output columns should be resolvable.")
- }
- }
-
- val projectFieldIdsIdx = projectFieldIdsToPoolIndex.getOrElseUpdate(
- projectFieldIds, {
- val idx = projectFieldIdsToPoolIndex.size
- val listBuilder = OperatorOuterClass.ProjectFieldIdList.newBuilder()
- projectFieldIds.foreach(id => listBuilder.addFieldIds(id))
- commonBuilder.addProjectFieldIdsPool(listBuilder.build())
- idx
- })
- taskBuilder.setProjectFieldIdsIdx(projectFieldIdsIdx)
+ // One task schema and one projection for the whole scan, both at pool index 0.
+ taskBuilder.setSchemaIdx(0)
+ taskBuilder.setProjectFieldIdsIdx(0)
val deleteFilesList =
extractDeleteFilesList(
@@ -1306,10 +1243,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
// Log deduplication summary
val allPoolSizes = Seq(
- schemaToPoolIndex.size,
+ commonBuilder.getSchemaPoolCount,
partitionSpecToPoolIndex.size,
nameMappingToPoolIndex.size,
- projectFieldIdsToPoolIndex.size,
+ commonBuilder.getProjectFieldIdsPoolCount,
partitionDataToPoolIndex.size,
deleteFileToPoolIndex.size,
deleteFilesToPoolIndex.size,
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
index 42220ea9b2..001ab5d721 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
@@ -95,8 +95,6 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
// scheme reads its bucket from the URL host (`requireSupportedStorageScheme`).
private val LocalStorageSchemes: Set[String] = Set("file", "memory")
private val MaxSupportedFormatVersion = 3
- // The Iceberg spec reserves field ids above `Integer.MAX_VALUE - 200` for metadata columns.
- private val MaxDataFieldId = Int.MaxValue - 200
private val ParquetWritePropertyPrefix = "write.parquet."
private val ParquetMrPropertyPrefix = "parquet."
private val CometS3CredentialProviderClassProperty =
@@ -508,7 +506,7 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
case None => Some("could not resolve the write schema's field ids")
case Some(fields) =>
fields.collectFirst {
- case (name, id) if id > MaxDataFieldId =>
+ case (name, id) if id > IcebergReflection.MaxDataFieldId =>
s"write schema includes metadata column $name, which iceberg-java fills with row " +
"lineage and the native writer does not write"
}
diff --git a/spark/src/test/resources/sql-tests/iceberg/nested_schema_pruning.sql b/spark/src/test/resources/sql-tests/iceberg/nested_schema_pruning.sql
new file mode 100644
index 0000000000..0488b6b26a
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/iceberg/nested_schema_pruning.sql
@@ -0,0 +1,136 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- The native Iceberg scan reads only the nested fields that Spark's nested schema pruning keeps.
+-- These queries check the pruned reads against Spark. CometIcebergNativeSuite checks that they
+-- scan fewer bytes, and covers deletes, time travel, and files without field ids.
+
+-- No Iceberg spark-runtime is published for Spark 4.2 yet, and the 4.0 runtime the build reuses
+-- is binary-incompatible with it. See https://github.com/apache/datafusion-comet/issues/4969.
+-- MaxSparkVersion: 4.1
+
+-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog
+-- Config: spark.sql.catalog.test_cat.type=hadoop
+-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test
+-- Config: spark.comet.scan.icebergNative.enabled=true
+
+statement
+DROP TABLE IF EXISTS test_cat.db.nested_pruning
+
+statement
+CREATE TABLE test_cat.db.nested_pruning (
+ id INT,
+ s STRUCT<a: INT, pad: STRING, inner: STRUCT<b: INT, pad: STRING>>,
+ items ARRAY<STRUCT<x: INT, pad: STRING>>,
+ m MAP<STRING, STRUCT<v: INT, pad: STRING>>
+) USING iceberg
+
+-- NULL structs, lists, maps, list elements, map values, and fields, so that the pruned read has
+-- to rebuild the validity of every level from the leaves it keeps. Empty lists and maps are valid
+-- with no child rows, which is another case.
+statement
+INSERT INTO test_cat.db.nested_pruning
+SELECT
+ CAST(id AS INT),
+ IF(id % 7 = 0, NULL, named_struct(
+ 'a', IF(id % 5 = 0, NULL, CAST(id AS INT)),
+ 'pad', repeat('p', 20),
+ 'inner', IF(id % 3 = 0, NULL, named_struct('b', CAST(id * 2 AS INT), 'pad', repeat('q', 20))))),
+ CASE WHEN id % 7 = 0 THEN NULL WHEN id % 11 = 0 THEN array() ELSE array(
+ named_struct('x', CAST(id AS INT), 'pad', repeat('r', 20)),
+ IF(id % 5 = 0, NULL, named_struct('x', CAST(-id AS INT), 'pad', repeat('s', 20)))) END,
+ CASE WHEN id % 7 = 0 THEN NULL WHEN id % 11 = 0 THEN map() ELSE map('k', IF(id % 3 = 0, NULL,
+ named_struct('v', CAST(id AS INT), 'pad', repeat('t', 20)))) END
+FROM range(100)
+
+query
+SELECT id, s.a FROM test_cat.db.nested_pruning ORDER BY id
+
+-- `s IS NULL` reads the validity of `s` pruned to `s.inner.b`.
+query
+SELECT id, s.inner.b, s IS NULL FROM test_cat.db.nested_pruning ORDER BY id
+
+query
+SELECT id, items.x FROM test_cat.db.nested_pruning ORDER BY id
+
+query
+SELECT id, m['k'].v FROM test_cat.db.nested_pruning ORDER BY id
+
+query
+SELECT id FROM test_cat.db.nested_pruning WHERE s.inner.b > 100 ORDER BY id
+
+-- The whole struct, so nothing is pruned.
+query
+SELECT id, s FROM test_cat.db.nested_pruning ORDER BY id
+
+statement
+DROP TABLE test_cat.db.nested_pruning
+
+-- A partition source the query does not project is appended to the task schema at its top level.
+-- `s.region` is nested, so a query that prunes it away reads with the full table schema instead,
+-- where an appended `region` would clash with the top-level one.
+statement
+DROP TABLE IF EXISTS test_cat.db.nested_pruning_nested_source
+
+statement
+CREATE TABLE test_cat.db.nested_pruning_nested_source (
+ id INT, region STRING, s STRUCT<region: STRING, a: INT, pad: STRING>
+) USING iceberg PARTITIONED BY (s.region)
+
+statement
+INSERT INTO test_cat.db.nested_pruning_nested_source
+SELECT CAST(id AS INT), 'top', named_struct(
+ 'region', IF(id % 2 = 0, 'east', 'west'), 'a', CAST(id AS INT), 'pad', repeat('p', 20))
+FROM range(100)
+
+query
+SELECT id, region, s.a FROM test_cat.db.nested_pruning_nested_source ORDER BY id
+
+query
+SELECT id, s.region FROM test_cat.db.nested_pruning_nested_source ORDER BY id
+
+statement
+DROP TABLE test_cat.db.nested_pruning_nested_source
+
+-- The partition source `p` is renamed to `q`, and a new `p` is added. A query that does not
+-- project `q` appends it to the task schema under its current name, apart from the new `p`. A
+-- bucket partition, because Iceberg rejects a new column named like an identity partition field.
+statement
+DROP TABLE IF EXISTS test_cat.db.nested_pruning_renamed_source
+
+statement
+CREATE TABLE test_cat.db.nested_pruning_renamed_source (
+ id INT, p STRING, s STRUCT<a: INT, pad: STRING>
+) USING iceberg PARTITIONED BY (bucket(4, p))
+
+statement
+INSERT INTO test_cat.db.nested_pruning_renamed_source
+SELECT CAST(id AS INT), CAST(id AS STRING), named_struct(
+ 'a', CAST(id AS INT), 'pad', repeat('p', 20))
+FROM range(100)
+
+statement
+ALTER TABLE test_cat.db.nested_pruning_renamed_source RENAME COLUMN p TO q
+
+statement
+ALTER TABLE test_cat.db.nested_pruning_renamed_source ADD COLUMN p STRING
+
+query
+SELECT id, p, s.a FROM test_cat.db.nested_pruning_renamed_source ORDER BY id
+
+statement
+DROP TABLE test_cat.db.nested_pruning_renamed_source
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index b00f3353e4..7a44594316 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -87,6 +87,20 @@ class CometIcebergNativeSuite
assertSingleNativeScan(cometPlan)
}
+ /**
+ * [[checkIcebergNativeScan]] with nested schema pruning off, so the scan reads its full schema.
+ */
+ private def checkFullSchemaNativeScan(query: String): Unit =
+ withSQLConf(CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") {
+ checkIcebergNativeScan(query)
+ }
+
+ /** [[checkIcebergNativeScan]] with nested schema pruning on, as by default, and then off. */
+ private def checkPrunedAndFullReads(query: String): Unit = {
+ checkIcebergNativeScan(query)
+ checkFullSchemaNativeScan(query)
+ }
+
/**
* Asserts a query stays native even when its DPP filter falls back (SMJ / non-Comet BHJ / reuse
* disabled), where only the runtime filter, not the scan, reverts to Spark.
@@ -1101,6 +1115,16 @@ class CometIcebergNativeSuite
tableName: String,
column: String,
value: Any,
+ warehouseDir: File): Unit =
+ commitEqualityDeletes(catalogName, namespace, tableName, column, Seq(value), warehouseDir)
+
+ /** Like [[commitEqualityDelete]], with one delete file holding every value in `values`. */
+ private def commitEqualityDeletes(
+ catalogName: String,
+ namespace: String,
+ tableName: String,
+ column: String,
+ values: Seq[Any],
warehouseDir: File): Unit = {
val catalog = spark.sessionState.catalogManager
.catalog(catalogName)
@@ -1112,9 +1136,12 @@ class CometIcebergNativeSuite
.table()
val deleteRowSchema = table.schema().select(column)
- val record: org.apache.iceberg.data.Record =
- org.apache.iceberg.data.GenericRecord.create(deleteRowSchema)
- record.setField(column, value)
+ val records = values.map { value =>
+ val record: org.apache.iceberg.data.Record =
+ org.apache.iceberg.data.GenericRecord.create(deleteRowSchema)
+ record.setField(column, value)
+ record
+ }
val out = table
.io()
.newOutputFile(new File(warehouseDir, s"eq-delete-${System.nanoTime()}.parquet").toString)
@@ -1122,11 +1149,16 @@ class CometIcebergNativeSuite
table,
out,
null,
- java.util.Collections.singletonList(record),
+ records.asJava,
deleteRowSchema)
table.newRowDelta().addDeletes(deleteFile).commit()
}
+ private def latestSnapshotId(table: String): Long = spark
+ .sql(s"SELECT snapshot_id FROM $table.snapshots ORDER BY committed_at DESC LIMIT 1")
+ .collect()(0)
+ .getLong(0)
+
// Regression for the native scan applying an equality delete keyed on a column that has since
// been dropped (schema evolution). The delete's field id is absent from the current schema, so
// the serde must recover it from the table's schema history to build the task schema; before the
@@ -1803,14 +1835,16 @@ class CometIcebergNativeSuite
checkNestedFieldEvolutionFallback(
s"SELECT id FROM $table WHERE s IS NULL ORDER BY id",
"s.b (added)")
- // The native read asks for the column's full nested type even when Spark prunes it.
+ // 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(
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 native read uses the current schema.
+ // 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)")
@@ -1818,7 +1852,51 @@ class CometIcebergNativeSuite
// A read that does not project the evolved column stays native.
checkIcebergNativeScan(s"SELECT id FROM $table ORDER BY id")
- spark.sql(s"DROP TABLE $table")
+ // 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.
+ val historical = "test_cat.db.evo_struct_historical"
+ spark.sql(s"""
+ CREATE TABLE $historical (id INT, c INT, s STRUCT<k: STRING, a: INT>)
+ USING iceberg PARTITIONED BY (s.k)
+ """)
+ spark.sql(s"INSERT INTO $historical VALUES (1, 100, named_struct('k', 'east', 'a', 10))")
+ spark.sql(s"ALTER TABLE $historical ADD COLUMN s.b INT")
+ spark.sql(s"""
+ INSERT INTO $historical VALUES (2, 200, named_struct('k', 'east', 'a', 20, 'b', 99))
+ """)
+ 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)")
+
+ // 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.
+ val unpartitioned = "test_cat.db.evo_struct_historical_unpartitioned"
+ spark.sql(s"CREATE TABLE $unpartitioned (id INT, c INT, s STRUCT<a: INT>) USING iceberg")
+ spark.sql(s"INSERT INTO $unpartitioned VALUES (1, 100, named_struct('a', 10))")
+ spark.sql(s"ALTER TABLE $unpartitioned ADD COLUMN s.b INT")
+ spark.sql(s"INSERT INTO $unpartitioned VALUES (2, 200, named_struct('a', 20, 'b', 99))")
+ val unpartitionedSnapshot = latestSnapshotId(unpartitioned)
+ spark.sql(s"ALTER TABLE $unpartitioned DROP COLUMN c")
+ spark.sql(s"ALTER TABLE $unpartitioned DROP COLUMN s.b")
+ val unpartitionedQuery =
+ 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)")
+ }
+ // 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.
+ val planned = spark.sql(unpartitionedQuery)
+ assertSingleNativeScan(planned.queryExecution.executedPlan)
+ withSQLConf(CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") {
+ checkAnswer(planned, Seq(Row(1, 100, 10), Row(2, 200, 20)))
+ }
+
+ Seq(table, historical, unpartitioned).foreach(t => spark.sql(s"DROP TABLE $t"))
}
}
}
@@ -2286,6 +2364,404 @@ class CometIcebergNativeSuite
}
}
+ // Spark's nested schema pruning reaches the native scan through the scan schema, so only the
+ // nested fields a query uses are read and the wide `pad` strings beside them are skipped.
+ // sql-tests/iceberg/nested_schema_pruning.sql checks the results. One data file holds every
+ // row, so each `pad` column chunk is larger than iceberg-rust's 1 MiB read coalescing, which
+ // would otherwise merge the reads of the kept chunks across the skipped ones.
+ test("nested schema pruning reads only the nested fields the query uses") {
+ 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_pruning"
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id INT,
+ s STRUCT<a: INT, pad: STRING>,
+ items ARRAY<STRUCT<x: INT, pad: STRING>>,
+ m MAP<STRING, STRUCT<v: INT, pad: STRING>>
+ ) USING iceberg
+ """)
+ spark.sql(s"""
+ INSERT INTO $table
+ SELECT
+ CAST(id AS INT),
+ named_struct('a', CAST(id AS INT), 'pad', pad),
+ array(named_struct('x', CAST(id AS INT), 'pad', pad)),
+ map('k', named_struct('v', CAST(id AS INT), 'pad', pad))
+ FROM (
+ SELECT id, concat_ws('', transform(array('a', 'b', 'c', 'd'),
+ salt -> sha2(concat(CAST(id AS STRING), salt), 256))) AS pad
+ FROM range(0, 20000, 1, 1))
+ """)
+
+ def bytesScanned(pruneNestedFields: Boolean): Long = {
+ var bytes = 0L
+ withSQLConf(
+ CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key ->
+ pruneNestedFields.toString) {
+ val df = spark.sql(s"SELECT sum(s.a), count(items.x), count(m['k'].v) FROM $table")
+ df.collect()
+ val scans = collectIcebergNativeScans(df.queryExecution.executedPlan)
+ assert(scans.length == 1, s"expected one native scan, got ${scans.length}")
+ bytes = scans.head.metrics("bytes_scanned").value
+ }
+ bytes
+ }
+ val prunedBytes = bytesScanned(pruneNestedFields = true)
+ val fullBytes = bytesScanned(pruneNestedFields = false)
+ assert(
+ prunedBytes * 4 < fullBytes,
+ s"pruned read should skip the pad fields: pruned=$prunedBytes, full=$fullBytes")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+
+ // A pruned task schema still needs the columns iceberg-rust uses beyond the projection: the
+ // partition source and the equality-delete key when the query projects neither. Tasks with
+ // deletes read with the pruned schema too, and every task of a scan reads with the same one.
+ // VERSION AS OF reads of columns dropped, renamed, or swapped after the snapshot use the
+ // snapshot's names.
+ test("nested schema pruning with deletes, partitions, and time travel") {
+ 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") {
+
+ // Also checks that every task, tasks with deletes included, reads with one schema that
+ // leaves out the pruned `pad` fields, and that the full schema read with pruning off gives
+ // the same answer.
+ def checkPrunedNativeScan(query: String): Unit = {
+ val (_, cometPlan) = checkSparkAnswer(query)
+ val scans = collectIcebergNativeScans(cometPlan)
+ assert(scans.length == 1, s"expected one native scan, got ${scans.length}\n$cometPlan")
+ // Planning commonData leaks manifest streams on Iceberg versions before 1.8.0.
+ if (icebergVersionAtLeast(1, 8)) {
+ val schemas = OperatorOuterClass.IcebergScanCommon
+ .parseFrom(scans.head.commonData)
+ .getSchemaPoolList
+ .asScala
+ assert(
+ schemas.length == 1 && !schemas.head.contains("\"pad\""),
+ s"$query: expected one task schema without pad, got:\n${schemas.mkString("\n")}")
+ }
+ checkFullSchemaNativeScan(query)
+ }
+
+ val morProperties = """
+ TBLPROPERTIES (
+ 'format-version' = '2',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read')
+ """
+ val rows = """
+ SELECT CAST(id AS INT) AS id, IF(id % 2 = 0, 'even', 'odd') AS p,
+ named_struct('a', CAST(id AS INT), 'pad', repeat('x', 100)) AS s
+ FROM range(200)
+ """
+
+ val mor = "test_cat.db.nested_pruning_mor"
+ spark.sql(s"""
+ CREATE TABLE $mor (
+ id INT, c STRING, r STRING, x INT, y INT, s STRUCT<a: INT, pad: STRING>)
+ USING iceberg $morProperties
+ """)
+ spark.sql(s"""
+ INSERT INTO $mor
+ SELECT id, p, concat('r', CAST(id AS STRING)), id, -id, s FROM ($rows)
+ """)
+ spark.sql(s"DELETE FROM $mor WHERE id % 10 = 0")
+ val snapshotWithDeletes = latestSnapshotId(mor)
+ commitEqualityDelete("test_cat", "db", "nested_pruning_mor", "id", 7, warehouseDir)
+ // Drop `c`, rename `r`, and swap the names of `x` and `y`.
+ Seq(
+ "DROP COLUMN c",
+ "RENAME COLUMN r TO r2",
+ "RENAME COLUMN x TO tmp",
+ "RENAME COLUMN y TO x",
+ "RENAME COLUMN tmp TO y").foreach(change => spark.sql(s"ALTER TABLE $mor $change"))
+ checkPrunedNativeScan(s"SELECT id, s.a FROM $mor ORDER BY id")
+ // The equality-delete key `id` is not projected, so the task schema appends it.
+ checkPrunedNativeScan(s"SELECT s.a FROM $mor ORDER BY s.a")
+ // The current table schema lacks `c` and names `r`, `x`, and `y` differently, so these
+ // tasks with deletes must read with the scan schema, which keeps the snapshot's names.
+ checkPrunedNativeScan(
+ s"SELECT id, c, r, x, y, s.a FROM $mor VERSION AS OF $snapshotWithDeletes ORDER BY id")
+
+ val partitioned = "test_cat.db.nested_pruning_partitioned"
+ spark.sql(s"""
+ CREATE TABLE $partitioned (id INT, p STRING, c STRING, s STRUCT<a: INT, pad: STRING>)
+ USING iceberg PARTITIONED BY (p) $morProperties
+ """)
+ spark.sql(s"""
+ INSERT INTO $partitioned
+ SELECT id, p, concat('c', CAST(id AS STRING)), s FROM ($rows)
+ """)
+ spark.sql(s"DELETE FROM $partitioned WHERE id % 10 = 0")
+ val partitionedSnapshot = latestSnapshotId(partitioned)
+ // The partition source `p` is not projected, so the task schema appends it.
+ checkPrunedNativeScan(s"SELECT id, s.a FROM $partitioned ORDER BY id")
+ checkPrunedNativeScan(s"SELECT p, count(s.a) FROM $partitioned GROUP BY p ORDER BY p")
+ // Once `p` takes the dropped `c`'s name, the task schema keeps the snapshot's names: `c`
+ // for the dropped column, and `p` for the partition source.
+ spark.sql(s"ALTER TABLE $partitioned DROP COLUMN c")
+ spark.sql(s"ALTER TABLE $partitioned RENAME COLUMN p TO c")
+ checkPrunedNativeScan(
+ s"SELECT id, c, s.a FROM $partitioned VERSION AS OF $partitionedSnapshot ORDER BY id")
+
+ // Two partition sources renamed into each other's names: `q` takes the dropped `c`'s name
+ // and `p` takes `q`'s. The task schema names both as the snapshot does, `p` and `q`.
+ val twoSources = "test_cat.db.nested_pruning_two_sources"
+ spark.sql(s"""
+ CREATE TABLE $twoSources (id INT, p INT, q INT, c INT, s STRUCT<a: INT, pad: STRING>)
+ USING iceberg PARTITIONED BY (bucket(4, p), bucket(4, q))
+ """)
+ spark.sql(s"INSERT INTO $twoSources SELECT id, id % 3, id % 5, id * 10, s FROM ($rows)")
+ val twoSourcesSnapshot = latestSnapshotId(twoSources)
+ Seq("DROP COLUMN c", "RENAME COLUMN q TO c", "RENAME COLUMN p TO q").foreach { change =>
+ spark.sql(s"ALTER TABLE $twoSources $change")
+ }
+ checkPrunedNativeScan(
+ s"SELECT id, c, s.a FROM $twoSources VERSION AS OF $twoSourcesSnapshot ORDER BY id")
+
+ // VERSION AS OF reads across type promotions. The batch adaptation after iceberg-rust
+ // cannot narrow a promoted type back, so the full schema must have the snapshot's types,
+ // not just be the newest schema that still has the dropped `c`. Each query reads one
+ // promoted column, so another promotion cannot steer the choice. The pruned schema cannot
+ // hold the nested partition source `s.k`, so the partitioned table reads the full schema
+ // with pruning on too.
+ Seq("promoted" -> "", "promoted_partitioned" -> "PARTITIONED BY (s.k)").foreach {
+ case (name, partitioning) =>
+ val promoted = s"test_cat.db.nested_pruning_$name"
+ spark.sql(s"""
+ CREATE TABLE $promoted (
+ id INT, x INT, d DECIMAL(9, 2), c INT, s STRUCT<k: STRING, a: INT>)
+ USING iceberg $partitioning
+ """)
+ spark.sql(s"""
+ INSERT INTO $promoted VALUES
+ (1, 10, 1.25, 100, named_struct('k', 'east', 'a', 1)),
+ (2, NULL, NULL, NULL, named_struct('k', 'west', 'a', CAST(NULL AS INT)))
+ """)
+ val snapshot = latestSnapshotId(promoted)
+ Seq(
+ "ALTER COLUMN x TYPE BIGINT",
+ "ALTER COLUMN d TYPE DECIMAL(18, 2)",
+ "ALTER COLUMN s.a TYPE BIGINT",
+ "DROP COLUMN c").foreach(change => spark.sql(s"ALTER TABLE $promoted $change"))
+ Seq("x", "d", "s.a").foreach { column =>
+ checkPrunedAndFullReads(
+ s"SELECT id, $column, c FROM $promoted VERSION AS OF $snapshot ORDER BY id")
+ }
+ spark.sql(s"DROP TABLE $promoted")
+ }
+
+ Seq(mor, partitioned, twoSources).foreach(t => spark.sql(s"DROP TABLE $t"))
+ }
+ }
+ }
+
+ // iceberg-rust builds an equality-delete predicate once per delete file, with the names of the
+ // task that loads it first, and binds it by name in every task that shares the reader, so every
+ // task must give a key the same name. Adaptive split sizing is off, so the files of each table
+ // land in one task group and share a reader.
+ test("nested schema pruning gives each equality-delete key one name in every task") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ // Equality deletes keyed on a since-dropped column need Iceberg 1.11's schema-history lookup.
+ assume(
+ icebergVersionAtLeast(1, 11),
+ "Equality delete on a dropped column requires Iceberg 1.11+ schema-history resolution")
+
+ 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 properties = """
+ TBLPROPERTIES ('format-version' = '2', 'read.split.adaptive-size.enabled' = 'false')
+ """
+
+ // A delete keyed on `y` covers only the second file. After `y` is dropped and `x0` is
+ // renamed to `x`, a delete keyed on `x` covers both, so one task needs `x` and the other
+ // needs `x` and `y`.
+ val renamed = "test_cat.db.key_names_renamed"
+ spark.sql(s"CREATE TABLE $renamed (id INT, x0 INT, y INT) USING iceberg $properties")
+ spark.sql(s"INSERT INTO $renamed VALUES (1, 10, 100), (2, 20, 100)")
+ spark.sql(s"INSERT INTO $renamed VALUES (3, 25, 200), (4, 30, 300)")
+ commitEqualityDelete("test_cat", "db", "key_names_renamed", "y", 200, warehouseDir)
+ spark.sql(s"ALTER TABLE $renamed DROP COLUMN y")
+ spark.sql(s"ALTER TABLE $renamed RENAME COLUMN x0 TO x")
+ commitEqualityDeletes(
+ "test_cat",
+ "db",
+ "key_names_renamed",
+ "x",
+ Seq(10, 25),
+ warehouseDir)
+ checkPrunedAndFullReads(s"SELECT id FROM $renamed ORDER BY id")
+
+ // Keys from schemas that never held both: `p` is dropped before `q` is added, and a delete
+ // keyed on each applies to the one data file.
+ val generations = "test_cat.db.key_names_generations"
+ spark.sql(s"CREATE TABLE $generations (id INT, p INT) USING iceberg $properties")
+ spark.sql(s"INSERT INTO $generations VALUES (1, 10), (2, 20), (3, 30)")
+ commitEqualityDelete("test_cat", "db", "key_names_generations", "p", 10, warehouseDir)
+ spark.sql(s"ALTER TABLE $generations DROP COLUMN p")
+ spark.sql(s"ALTER TABLE $generations ADD COLUMN q INT")
+ commitEqualityDelete("test_cat", "db", "key_names_generations", "q", 99, warehouseDir)
+ checkPrunedAndFullReads(s"SELECT id FROM $generations ORDER BY id")
+
+ // The first spec's files need the nested partition source `s.region`, which the query
+ // prunes away, so this VERSION AS OF read uses the full schema. A global delete keyed on
+ // `k` covers files of both specs, and `k` and `j` swap names after the snapshot, so the
+ // full schema must carry the snapshot's names, as the pruned one does.
+ val swapped = "test_cat.db.key_names_swapped"
+ spark.sql(s"""
+ CREATE TABLE $swapped (id INT, k INT, j INT, s STRUCT<region: STRING, a: INT>)
+ USING iceberg PARTITIONED BY (s.region) $properties
+ """)
+ spark.sql(s"""
+ INSERT INTO $swapped VALUES
+ (1, 10, 100, named_struct('region', 'east', 'a', 1)),
+ (2, 20, 200, named_struct('region', 'west', 'a', 2))
+ """)
+ loadIcebergTable(spark, "test_cat", "db", "key_names_swapped")
+ .asInstanceOf[org.apache.iceberg.Table]
+ .updateSpec()
+ .removeField("s.region")
+ .commit()
+ spark.sql(s"""
+ INSERT INTO $swapped VALUES
+ (3, 10, 300, named_struct('region', 'east', 'a', 3)),
+ (4, 30, 400, named_struct('region', 'west', 'a', 4))
+ """)
+ commitEqualityDelete("test_cat", "db", "key_names_swapped", "k", 10, warehouseDir)
+ val snapshot = latestSnapshotId(swapped)
+ Seq("RENAME COLUMN k TO tmp", "RENAME COLUMN j TO k", "RENAME COLUMN tmp TO j").foreach {
+ change => spark.sql(s"ALTER TABLE $swapped $change")
+ }
+ checkPrunedAndFullReads(s"SELECT id, k FROM $swapped VERSION AS OF $snapshot ORDER BY id")
+
+ Seq(renamed, generations, swapped).foreach(t => spark.sql(s"DROP TABLE $t"))
+ }
+ }
+ }
+
+ // Data files written without Iceberg field ids and read with no name mapping: iceberg-rust then
+ // projects whole top-level columns by position, and a by-name struct cast narrows each one to
+ // the pruned type. Spark's own reader returns NULL nested fields for such a table, so the pruned
+ // read is checked against the native read of every nested field instead.
+ test("nested schema pruning on data files without field ids") {
+ 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 dataPath = s"${warehouseDir.getAbsolutePath}/nested_no_ids"
+ spark
+ .sql("""
+ SELECT
+ CAST(id AS INT) AS id,
+ named_struct('a', CAST(id AS INT), 'pad', repeat('x', 50), 'b', CAST(-id AS INT))
+ AS s,
+ array(named_struct('x', CAST(id AS INT), 'pad', repeat('y', 50))) AS items
+ FROM range(100)
+ """)
+ .coalesce(1)
+ .write
+ .parquet(dataPath)
+ spark.sql("CREATE NAMESPACE IF NOT EXISTS test_cat.db")
+ val table = "test_cat.db.nested_no_ids"
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id INT,
+ s STRUCT<a: INT, pad: STRING, b: INT>,
+ items ARRAY<STRUCT<x: INT, pad: STRING>>
+ ) USING iceberg
+ """)
+ try {
+ val tableUtilClass = Class.forName("org.apache.iceberg.spark.SparkTableUtil")
+ val icebergTable = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[org.apache.iceberg.spark.SparkCatalog]
+ .loadTable(
+ org.apache.spark.sql.connector.catalog.Identifier.of(Array("db"), "nested_no_ids"))
+ .asInstanceOf[org.apache.iceberg.spark.source.SparkTable]
+ .table()
+ spark.sql(s"CREATE TABLE parquet_no_ids USING parquet LOCATION '$dataPath'")
+ tableUtilClass
+ .getMethod(
+ "importSparkTable",
+ classOf[org.apache.spark.sql.SparkSession],
+ classOf[org.apache.spark.sql.catalyst.TableIdentifier],
+ classOf[org.apache.iceberg.Table],
+ classOf[String])
+ .invoke(
+ null,
+ spark,
+ new org.apache.spark.sql.catalyst.TableIdentifier("parquet_no_ids"),
+ icebergTable,
+ s"${warehouseDir.getAbsolutePath}/staging")
+ spark.sql(
+ s"ALTER TABLE $table UNSET TBLPROPERTIES IF EXISTS ('schema.name-mapping.default')")
+
+ def nativeRows(query: String, pruneNestedFields: Boolean): Seq[Row] = {
+ var rows = Seq.empty[Row]
+ withSQLConf(
+ CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key ->
+ pruneNestedFields.toString) {
+ val df = spark.sql(query)
+ rows = df.collect().toSeq
+ assertSingleNativeScan(df.queryExecution.executedPlan)
+ }
+ rows
+ }
+ // `s.b` is the struct's last field, so a positional cast would read `s.a` instead.
+ 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)
+ }
+
+ spark.sql(s"DROP TABLE $table")
+ spark.sql("DROP TABLE parquet_no_ids")
+ } catch {
+ case _: ClassNotFoundException =>
+ cancel("SparkTableUtil not available")
+ }
+ }
+ }
+ }
+
test("UUID type - native Iceberg UUID column (reproduces type mismatch)") {
assume(icebergAvailable, "Iceberg not available in classpath")
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergReadBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergReadBenchmark.scala
index 1a171a5ada..ab5449bcc3 100644
--- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergReadBenchmark.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergReadBenchmark.scala
@@ -19,7 +19,10 @@
package org.apache.spark.sql.benchmark
+import java.io.File
+
import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.comet.CometIcebergNativeScanExec
import org.apache.spark.sql.types._
import org.apache.comet.CometConf
@@ -28,7 +31,8 @@ import org.apache.comet.CometConf
* Benchmark to measure Comet Iceberg read performance. To run this benchmark:
* `SPARK_GENERATE_BENCHMARK_FILES=1 make
* benchmark-org.apache.spark.sql.benchmark.CometIcebergReadBenchmark` Results will be written to
- * "spark/benchmarks/CometIcebergReadBenchmark-**results.txt".
+ * "spark/benchmarks/CometIcebergReadBenchmark-**results.txt". Pass `--nested-only` to run only
+ * the nested column cases.
*/
object CometIcebergReadBenchmark extends CometBenchmarkBase {
@@ -75,10 +79,107 @@ object CometIcebergReadBenchmark extends CometBenchmarkBase {
}
}
+ /**
+ * Reads a TPC-H style nested `partsupp` table, whose `partsupp_data` column is an
+ * `array<struct>` dominated by a text field, as in the nested TPC-H schema. Spark prunes the
+ * struct to the fields a query uses. The pruned Comet case passes that pruned schema to
+ * iceberg-rust, and the other reads every nested field and drops the unused ones after
+ * decoding.
+ */
+ def icebergNestedProjectionBenchmark(values: Int): Unit = {
+ withTempPath { dir =>
+ configureIcebergHadoopCatalog(new File(dir, "iceberg-warehouse"))
+ val table = s"$defaultIcebergCatalog.db.partsupp_nested"
+ spark.sql(s"""
+ CREATE TABLE $table (
+ ps_partkey BIGINT,
+ ps_suppkey BIGINT,
+ partsupp_data ARRAY<STRUCT<
+ ps_availqty: INT, ps_supplycost: DECIMAL(15, 2), ps_comment: STRING>>
+ ) USING iceberg
+ TBLPROPERTIES ('format-version' = '2', 'write.parquet.compression-codec' = 'zstd')
+ """)
+ // TPC-H style comments: 16 words from a small vocabulary, about 128 characters each.
+ val words = ("furiously carefully quickly slyly blithely fluffily ironic final regular " +
+ "express special pending unusual bold even silent idle ruthless packages requests " +
+ "accounts deposits foxes ideas theodolites pinto beans instructions dependencies " +
+ "excuses platelets asymptotes courts dolphins sleep wake haggle nag use boost")
+ .split(" ")
+ .map(w => s"'$w'")
+ .mkString(", ")
+ spark.sql(s"""
+ INSERT INTO $table
+ SELECT id DIV 4, id % 4 + 1, array(named_struct(
+ 'ps_availqty', CAST(pmod(xxhash64(id, 1), 9999) + 1 AS INT),
+ 'ps_supplycost', CAST(pmod(xxhash64(id, 2), 99901) / 100 + 1 AS DECIMAL(15, 2)),
+ 'ps_comment', concat_ws(' ', transform(sequence(1, 16),
+ i -> element_at(array($words), CAST(pmod(xxhash64(id, i), 40) + 1 AS INT))))))
+ FROM range($values)
+ """)
+
+ val cometIceberg = Seq(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")
+ def pruneNestedFields(enabled: Boolean): (String, String) =
+ CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> enabled.toString
+
+ def explodeField(aggregate: String, field: String): String =
+ s"SELECT $aggregate FROM $table LATERAL VIEW explode(partsupp_data.$field) e AS v"
+
+ Seq(
+ "one small field, as TPC-H q20 reads partsupp" -> explodeField("sum(v)", "ps_availqty"),
+ "the text field, which pruning cannot skip" ->
+ explodeField("sum(length(v))", "ps_comment"))
+ .foreach { case (name, query) =>
+ val benchmark =
+ new Benchmark(s"Nested Iceberg column, $name", values.toLong, output = output)
+
+ benchmark.addCase("SQL Iceberg - Spark") { _ =>
+ spark.sql(query).noop()
+ }
+ benchmark.addCase("SQL Iceberg - Comet, every nested field") { _ =>
+ withSQLConf(cometIceberg :+ pruneNestedFields(false): _*) {
+ spark.sql(query).noop()
+ }
+ }
+ benchmark.addCase("SQL Iceberg - Comet, pruned nested fields") { _ =>
+ withSQLConf(cometIceberg :+ pruneNestedFields(true): _*) {
+ spark.sql(query).noop()
+ }
+ }
+
+ val bytesScanned = Seq(false, true).map { enabled =>
+ var bytes = 0L
+ withSQLConf(cometIceberg :+ pruneNestedFields(enabled): _*) {
+ val df = spark.sql(query)
+ df.collect()
+ bytes = collect(df.queryExecution.executedPlan) {
+ case scan: CometIcebergNativeScanExec => scan.metrics("bytes_scanned").value
+ }.sum
+ }
+ bytes / 1024.0 / 1024.0
+ }
+ warn(
+ benchmark,
+ f"Comet bytes scanned: every nested field ${bytesScanned.head}%.1f MiB, " +
+ f"pruned nested fields ${bytesScanned.last}%.1f MiB")
+ benchmark.run()
+ }
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+
override def runCometBenchmark(mainArgs: Array[String]): Unit = {
- runBenchmarkWithTable("SQL Single Numeric Iceberg Column Scan", 1024 * 1024 * 128) { v =>
- Seq(BooleanType, ByteType, ShortType, IntegerType, LongType, FloatType, DoubleType)
- .foreach(icebergScanBenchmark(v, _))
+ if (!mainArgs.contains("--nested-only")) {
+ runBenchmarkWithTable("SQL Single Numeric Iceberg Column Scan", 1024 * 1024 * 128) { v =>
+ Seq(BooleanType, ByteType, ShortType, IntegerType, LongType, FloatType, DoubleType)
+ .foreach(icebergScanBenchmark(v, _))
+ }
+ }
+ runBenchmark("SQL Nested Iceberg Column Scan") {
+ icebergNestedProjectionBenchmark(1024 * 1024 * 4)
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org