comphead commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4263202776
##########
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:
Confirmed from the code, thanks. Since 6213d0576e the full task schema can
be older than the table's, and the nested-evolution check walked only the table
and scan schemas. 8566834663 adds the full task schema to that check, so your
read falls back to Spark with `s.b (added)` instead of failing in the native
scan. The struct fallback test now has your table. Keeping this case native
would need the nested partition source merged back into its pruned struct,
which I'd leave for a follow-up.
--
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]
