comphead commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4265103364
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -774,15 +779,17 @@ case class CometScanRule(session: SparkSession)
// fields to the table's by field id (apache/iceberg-rust#2617). A
file written before a
// nested field was added (to a struct, or to a struct inside a list
or map) fails the
// native scan with "Incorrect number of arrays for StructArray
fields", and a nested field
- // renamed since the file was written can read back as NULL. The
native read asks for a
- // projected column's full nested type even when Spark prunes it,
taken from the current
- // table schema, or from the scan schema when VERSION AS OF reads a
dropped column (see
- // CometIcebergNativeScan). A FileScanTask does not record which
schema wrote its file, so
- // fall back when any schema in the table's history lacks a nested
field of a projected
- // column or names it differently.
+ // renamed since the file was written can read back as NULL. The
native read usually asks
+ // only for the nested fields Spark's schema pruning keeps, but a scan
can still read with
+ // its full task schema (see IcebergReflection.nativeTaskSchemas),
which for VERSION AS OF
+ // can be an older schema than the table's. So this check walks the
table, scan, and full
+ // task schemas, and falls back even when the changed field is pruned
away. A FileScanTask
+ // does not record which schema wrote its file, so fall back when any
schema in the table's
+ // history lacks a nested field of a projected column or names it
differently.
val nestedFieldsSupported =
try {
- val schemas = Seq(metadata.tableSchema, metadata.scanSchema)
+ val schemas =
+ Seq(metadata.tableSchema, metadata.scanSchema) ++
metadata.fullTaskSchema
Review Comment:
Agreed, thanks for running both variants. 5395e50ba3 adds
`CometIcebergNativeScanMetadata.taskSchema`, which picks the pruned task schema
while pruning is on and can apply, and the full one otherwise. `CometScanRule`
checks that schema, and `CometIcebergNativeScan` serializes it, so they cannot
drift. Your unpartitioned table now stays native with pruning on and falls back
with pruning off. Both are in the struct fallback test, next to the partitioned
case, which still falls back.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -1039,12 +1046,21 @@ case class CometScanRule(session: SparkSession)
}
}
+ // Every task needs each partition source and equality-delete key
under one name (see
+ // IcebergReflection.nativeTaskSchemas).
+ val taskSchemaNamesSupported = metadata.fullTaskSchema.isDefined
+ if (!taskSchemaNamesSupported) {
+ fallbackReasons += "Comet's native Iceberg reader cannot name a
partition source or " +
+ "equality-delete key the same way in every task: no schema in the
table's history " +
+ "names it apart from the columns the scan reads"
Review Comment:
Added to the Current limitations list in 5395e50ba3, after the nested schema
evolution entry, with your wording.
--
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]
