andygrove commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4273526402
##########
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:
This picks the full schema by top-level ids and names only. For `VERSION AS
OF`, that can be a schema newer than the snapshot that gives a projected column
a promoted type. The batch adaptation after iceberg-rust won't narrow that back
to the type Spark reads, so the native scan fails. For example:
```sql
CREATE TABLE t (id INT, x INT, c INT, s STRUCT<k: STRING, a: INT>) USING
iceberg PARTITIONED BY (s.k);
INSERT INTO t VALUES (1, 10, 100, named_struct('k', 'east', 'a', 1));
-- take snapshot S here
ALTER TABLE t ALTER COLUMN x TYPE BIGINT;
ALTER TABLE t DROP COLUMN c;
SELECT id, x, c FROM t VERSION AS OF S;
```
This fails with `Parquet column cannot be converted in file . Column: [[x]],
Expected: int, Found: INT64`, with pruning on or off. The pruned-away `s.k`
sends the scan to the full schema, and the newest schema that still has `c` is
the one right after the promotion. On `main` and `branch-1.1` the dropped `c`
makes the task read the scan schema, so the query matches Spark. Without the
nested partition source, the same read works with pruning on and now fails with
pruning off. Promoting a float to double, widening a decimal, or promoting a
nested field such as `s.a` fails the same way.
What do you think about matching types too when picking the full schema? A
candidate would qualify only when every field the scan reads, nested ones
included, has the same name and type in it. The snapshot's own schema always
qualifies, so this shouldn't add a fallback. It would also fix reads across a
promotion that fail on `main` today, such as the query above without the drop.
I filed #6993 for those, since `branch-1.1` has them too. Could the time travel
test cover the read above with pruning on and off, and the unpartitioned
version with pruning off?
--
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]
