andygrove commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4241759688
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -1130,49 +1162,43 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
// verbatim for iceberg-rust to decode. Unencrypted files
leave the field unset.
keyMetadataBytes(keyMetadataMethod,
dataFile).foreach(taskBuilder.setKeyMetadata)
- val taskSchema = taskSchemaMethod.invoke(task)
-
val deletes =
IcebergReflection.getDeleteFilesFromTask(task,
fileScanTaskClass)
val hasDeletes = !deletes.isEmpty
- val baseSchema: AnyRef =
+ // An equality delete may be keyed on a column dropped from
the current schema
+ // (schema evolution). iceberg-rust must read that column to
apply the delete,
+ // so union the equality-delete field ids into the task
schema, resolving any
+ // dropped ones from the table's schema history (mirrors
Iceberg-Java's
+ // DeleteFilter.fileProjection).
+ val equalityFieldIds =
if (hasDeletes) {
- // An equality delete may be keyed on a column dropped
from the current schema
- // (schema evolution). iceberg-rust must read that column
to apply the delete,
- // so union the equality-delete field ids into the task
schema, resolving any
- // dropped ones from the table's schema history (mirrors
Iceberg-Java's
- // DeleteFilter.fileProjection).
- val equalityFieldIds = deletes.asScala.flatMap { df =>
+ deletes.asScala.flatMap { df =>
requiredEqualityFieldIds(deleteFileClass,
df).asScala.map(_.intValue())
}.toSeq
- if (equalityFieldIds.nonEmpty) {
- IcebergReflection
- .schemaWithRequiredFields(taskSchema, metadata.table,
equalityFieldIds)
- .asInstanceOf[AnyRef]
- } else {
- taskSchema
- }
} else {
- if (hasHistoricalColumns) {
- metadata.scanSchema.asInstanceOf[AnyRef]
- } else {
- metadata.tableSchema.asInstanceOf[AnyRef]
- }
+ Nil
}
// iceberg-rust validates a FileScanTask by resolving its
partition spec against
// the task schema, so a task carrying a partition spec needs
that spec's source
// columns present even when the query projects them out (e.g.
selecting only
// _spec_id / _partition). Union them in. project_field_ids
still drives the read,
// so these columns are not materialized into the output.
- val schema: AnyRef =
- IcebergReflection
- .schemaWithRequiredFields(
- baseSchema,
- metadata.table,
- IcebergReflection.partitionSourceFieldIds(task,
fileScanTaskClass))
- .asInstanceOf[AnyRef]
+ val requiredFieldIds = equalityFieldIds ++
+ IcebergReflection.partitionSourceFieldIds(task,
fileScanTaskClass)
+
+ val baseSchema: AnyRef =
+ if (canUsePrunedSchema(requiredFieldIds)) {
Review Comment:
There's one more constraint on the task schemas. iceberg-rust shares
equality deletes across the tasks of a partition. Comet builds one
`ArrowReader` per partition, and its `DeleteFilter` builds each equality-delete
predicate once per delete file, with the column names of the task that loads it
first. Every other task then binds that predicate by name against its own
schema (`build_equality_delete_predicate` in `delete_filter.rs`). So every task
of a scan needs to give a key the same name. Two things here break that. I
turned off `read.split.adaptive-size.enabled` in both repros so the tasks share
one reader.
The first is that the appended names now depend on what else a task needs. I
tried an unpartitioned table `t(id, x0, y)` with two data files. An equality
delete on `y = 200` covers only the second file. Then I dropped `y`, renamed
`x0` to `x`, and committed an equality delete on `x` in `(10, 25)` that covers
both files. The first task appends `x` from the current schema. The second also
needs `y`, so it takes both fields from the oldest schema, where `x` is still
`x0`. `SELECT id FROM t ORDER BY id` fails with `Field x not found in schema`
with pruning on. Spark returns `3`, and Comet matches it with pruning off.
The second is the base schema in a time travel read. A task whose nested
partition source the query prunes keeps `task.schema()`, with the current
names, while the other tasks use the snapshot's names. I tried a table
partitioned by `s.region` that then moves to an unpartitioned spec, so one
global equality delete on `k = 10` covers files from both specs, and renamed
`k` after the snapshot. `SELECT id, k FROM t VERSION AS OF <s>` fails with
`Field k not found in schema`. When `k` and `j` swap names instead, it either
fails with `Field j not found in schema` or returns `[1,100] [4,30]` where
Spark returns `[2,30] [4,30]`, depending on which task loads the delete first.
In that last case a row the delete removes comes back and a live row
disappears. With pruning off, every task with deletes reads `task.schema()`.
That gives the #6822 wrong values but deletes the right rows.
What do you think about resolving each missing field id to one name for the
whole scan, rather than per task? It could start from the scan's own schema,
which is the snapshot's schema for `VERSION AS OF` and the current one
otherwise, and use the history only for ids that schema lacks. Using that
schema as the full-schema fallback too would cover the time travel case, and
#6822's full-schema half with it. Doing this during planning would also let a
case it can't resolve fall back to Spark instead of failing the query. Could
the tests cover both cases, with adaptive split sizing off so the tasks share a
reader?
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -741,20 +743,30 @@ 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)
+ val existingNames = existing.asScala.map(fieldName).toSet
+ // Taking every field from one schema keeps their names apart, which a
name picked per field
+ // cannot promise once columns have been renamed into each other's
names. A VERSION AS OF
+ // scan schema carries the snapshot's names, so a schema also must not
name a field like a
+ // column `baseSchema` has. The current schema comes first, to keep
current names, and then
+ // the others newest first, so that a promoted column keeps its widest
type.
+ val schemas = getMethod(table.getClass, "schema").invoke(table) +:
+ getAllSchemas(table).reverse
+ val resolvedFields = schemas.iterator
+ .map(schema => missingIds.flatMap(findFieldObject(schema, _)))
+ .find { fields =>
+ val names = fields.map(fieldName)
+ fields.length == missingIds.length && names.distinct.length ==
names.length &&
Review Comment:
I can confirm this one through Comet at d88698c442. With your table, `SELECT
id FROM t ORDER BY id` throws `Cannot resolve field ids 2,3 from one schema in
the table schema history under names the task schema does not use` with pruning
on. Spark returns `2, 3`. With pruning off it matches Spark, because the base
schema already has `q` and only `p` needs appending. My comment on
`CometIcebergNativeScan.scala` is related, since whatever replaces the
one-schema rule also has to give each key the same name in every task.
--
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]
