sunchao commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4238599043
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -741,28 +743,68 @@ object IcebergReflection extends Logging {
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)
+ // A field is appended under a name it has had that the task schema does
not use yet. The
+ // current schema comes first, to keep a live column's current name and
type, and then
+ // table.schemas(), oldest first. A VERSION AS OF scan schema carries
the snapshot's names,
+ // so a current name can clash with a column the snapshot still has.
+ val schemas = getMethod(table.getClass, "schema").invoke(table) +:
getAllSchemas(table)
+ val names =
scala.collection.mutable.Set(existing.asScala.map(fieldName).toSeq: _*)
+ missingIds.foreach { id =>
+ val (field, name) = schemas.iterator
+ .flatMap(findFieldObject(_, id))
+ .map(f => (f, fieldName(f)))
+ .find { case (_, n) => !names.contains(n) }
Review Comment:
[P2] Resolve missing fields as a consistent set before reserving their
names. For a snapshot containing `id, p, q, c`, partitioned by `bucket(4, p),
bucket(4, q)`, drop `c`, rename `q` to `c`, then rename `p` to `q`. A
historical `SELECT id, c` needs both partition sources appended. This loop
first assigns the current name `q` to the old `p` field. The old `q` field then
cannot use either its current name `c` or historical name `q`, and
serialization throws `Cannot resolve field id 3 in table schema history under
an unused name`. The valid historical names `p, q` would work. This additional
two-source case survives the single-source collision fix and breaks a read
supported by the base and release, even with pruning disabled. Prefer resolving
from the complete schema associated with the scan's snapshot before falling
back to history, and add this case to the time-travel test.
Evidence: Recompiled the exact `335e7e0c` helper and reproduced the failure
using real Iceberg `InMemoryCatalog` tables and `planFiles()` on 1.5.2, 1.8.1,
1.10.0, and 1.11.0. Planned partition source IDs are `[2,3]`; the historical
projection contains IDs `[1,4]`. The base helper succeeds with names
`id,c,p,q`, while HEAD throws for ID 3. Reversing the required IDs succeeds,
confirming order dependence. Because historical `c` was dropped and there are
no deletes, both pruning settings select the historical projection. SQL
trigger: create `t(id INT,p INT,q INT,c INT)` partitioned by
`bucket(4,p),bucket(4,q)`; insert `(1,10,20,30)` and record snapshot S; drop
`c`; rename `q` to `c`; rename `p` to `q`; query `SELECT id,c FROM t VERSION AS
OF S`, whose expected row is `(1,30)`. This SQL query was not run through a
rebuilt Comet. Harness and logs: `/tmp/pr6725-335e7e0c-verified/`.
--
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]
