comphead commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4274520574
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -707,89 +708,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) }
Review Comment:
Confirmed from the code, thanks. `nativeTaskSchemas` matched the full schema
on top-level ids and names only, so in your repro it took the schema right
after the promotion, the newest one that still has `c`. iceberg-rust then
returned `x` as `Int64`, which the adaptation in `iceberg_scan.rs` won't
narrow. `main` reads the scan schema there because of the dropped `c`.
86eaa60d37 does what you suggested. The full schema is now the newest schema
in which every field the scan reads, at any depth, has the same name and type.
Types are compared with Iceberg's `Type.equals`, so a decimal's precision
counts too. Reads without `VERSION AS OF` still take the current schema.
The time travel test now has your read on a table partitioned by `s.k` and
the unpartitioned version, each with pruning on and off. It also reads a wider
decimal and a promoted `s.a`, one promoted column per query, so one promotion
cannot pick the schema for another. Against the old choice, every read that
uses the full schema fails with `Parquet column cannot be converted`. A check
on type ids alone misses the decimal, and one on top-level columns alone misses
`s.a`.
Thanks for filing #6993. It stays open for `branch-1.1`. On this branch the
same rule should cover its cases too, though I haven't run its repros.
--
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]
