comphead commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4260769444
##########
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:
Confirmed, thanks both. The one-schema rule assumed a task's keys always
coexist in some schema, and an equality delete committed after `p` was dropped
and `q` added breaks that. 6213d0576e drops the rule. Each required id now gets
one field for the whole scan, from the scan's full schema when it has it, and
otherwise from the newest schema whose name for it no other column uses. Keys
from different generations no longer need to share a schema. The new test has
your table.
##########
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:
Thanks for tracing this into `DeleteFilter`. At Comet's pin,
`build_equality_delete_predicate` binds the cached predicate by name against
each task's schema, so the names have to agree. 6213d0576e follows your
suggestion. Every task of a scan reads with one schema, built in
`CometIcebergNativeScanMetadata.extract` during planning from the partition
sources and equality keys that `CometScanRule` collects. The names come from
the scan's full schema, the newest schema whose top-level columns match the
scan schema by id and name: the snapshot's names for `VERSION AS OF`, the
current ones otherwise. Iceberg exposes the snapshot schema itself only through
package-private `SparkBatchQueryScan` state, and any schema that matches the
scan's names gives every task the same names. Only an equality key dropped
before that schema comes from the history. The full schema is also what a scan
reads when a nested partition source is pruned away or the config is off, so
time travel uses the snapshot
's names there too, which covers the full-schema half of #6822. A case with no
free name falls back to Spark at planning. The new test covers both of your
cases with adaptive split sizing off, and sunchao's.
--
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]
