mbutrovich commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4261112163
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -707,89 +707,136 @@ 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]))
/**
- * 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.
+ * 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.
*
- * 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
+ * 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 names come from
the full schema, the
+ * newest schema, current first, that names every column of the scan schema
the same way. That
+ * is the current schema unless a VERSION AS OF snapshot named a column
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 scanColumns = columnNamesById(prunedScanSchema).toSet
+ val schemas = tableSchema +: getAllSchemas(table).reverse
+ val full = schemas.iterator
+ .map(schema => schema -> columnNamesById(schema))
+ .find { case (_, columns) => scanColumns.subsetOf(columns.toSet) }
+ full match {
+ case None => (None, None)
+ case Some((fullSchema, fullColumns)) =>
+ 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)))
Review Comment:
> Could the planning guard validate `metadata.fullTaskSchema` as well and
fall back for this case?
I ran this table through Comet. At 6213d0576e, `SELECT id, c, s.a FROM t
VERSION AS OF S` failed natively with pruning on and off. An unpartitioned
variant also failed with pruning off: `t(id INT, c INT, s STRUCT<a: INT>)`, add
`s.b`, write a second file, take snapshot `S`, then drop `c` and `s.b`. At
8566834663 both now fall back and match Spark. I left a related question on the
new check in `CometScanRule.scala`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
