sunchao commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4252902919
##########
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:
[P2] [P2] Validate the historical full task schema for nested evolution.
Create `t(id INT, c INT, s STRUCT<k: STRING, a: INT>) USING iceberg PARTITIONED
BY (s.k)`, write a file, add `s.b INT`, write another file, and record snapshot
S. Then drop `c` and `s.b`. `SELECT id, c, s.a FROM t VERSION AS OF S` selects
a historical full schema containing `s.b`, because `c` is historical and the
unprojected nested partition source `s.k` prevents pruning. However,
`CometScanRule` checks only `metadata.tableSchema` and `metadata.scanSchema`,
neither of which contains `s.b`. Native execution consequently fails reading
the older file with `Incorrect number of arrays for StructArray fields,
expected 3 got 2`, while Spark and the base schema succeed. Both pruning
settings select this unsafe full schema. Could the planning guard validate
`metadata.fullTaskSchema` as well and fall back for this case?
Evidence: Fresh reproduction in `/tmp/pr6725-6213-review-oct12-c2z7stms/`:
`NestedHistoryProbe.scala` writes `(1,100,{k:east,a:10})`, adds `s.b`, writes
`(2,200,{k:east,a:20,b:99})`, records S, then drops `c` and `s.b`. Spark
4.1.3/Iceberg 1.11.0 returns `[1,100,10]` and `[2,200,20]`. The exact-head
helpers produce `prunedTaskSchema=None`; the current guard reports no changes,
whereas checking the selected full schema reports `s.b (added)`. Reading those
same files with pinned iceberg-rust `af1da4c5bc86178c38c1c6db0578bdd9fb7021e3`
reports `BASE: SUCCESS rows=2` and `HEAD: ERROR ... expected 3 got 2`. The base
augmentation helper also matches the current `branch-1.1` implementation.
Sources, generated schemas, and logs are preserved in that directory.
--
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]
