andygrove opened a new issue, #7031:
URL: https://github.com/apache/datafusion-comet/issues/7031
### Describe the bug
On Spark 3.4 with `spark.sql.parquet.fieldId.read.enabled=true`, the native
Parquet scan returns NULL for the fields of a struct column that the read
schema renames and matches by field id, whenever Spark's nested schema pruning
prunes that struct. Spark 3.4 returns the values. Filters on those fields then
drop rows. The whole struct, read without pruning, is correct.
`SchemaPruning.pruneSchema` rebuilds a pruned root field as
`StructField(name, dataType, nullable)`, so the pruned `relation.dataSchema`
loses the root's `parquet.field.id` metadata on every Spark version, while the
nested fields keep theirs. Spark 3.4.3's reader works from
`FileSourceScanExec.requiredSchema`, which still carries the root id, so it
matches the renamed root by id. Spark 3.5.9 and 4.1.3 drop the root id from
`requiredSchema` too, so Spark itself returns NULL there, and Comet matches it.
`CometNativeScan` serializes the relation's data schema as the native table
schema, so the native adapter has no id for the root and looks it up by its new
name, which the file does not have.
### Steps to reproduce
In a `CometTestBase` suite on the Spark 3.4 profile:
```scala
def withId(f: StructField, id: Int) =
f.copy(metadata = new MetadataBuilder().putLong("parquet.field.id",
id).build())
def schema(root: String) = StructType(Seq(
withId(StructField("id", IntegerType), 1),
withId(StructField(root, StructType(Seq(
withId(StructField("a", IntegerType), 11),
withId(StructField("b", StringType), 12)))), 2)))
spark.createDataFrame(
spark.sparkContext.parallelize(Seq(Row(1, Row(10, "x")), Row(2, Row(20,
"y"))), 1),
schema("s"))
.write.parquet(path)
withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
val df = spark.read.schema(schema("s2")).parquet(path)
df.select("id", "s2").collect()
df.select("id", "s2.a").collect()
df.where("s2.a > 15").select("id").collect()
}
```
On `main` at 66c63bd307 with Spark 3.4.3:
| Query | Spark | Comet native scan |
| --- | --- | --- |
| `select("id", "s2")` | `[1,[10,x]] [2,[20,y]]` | `[1,[10,x]] [2,[20,y]]` |
| `select("id", "s2.a")` | `[1,10] [2,20]` | `[1,null] [2,null]` |
| `where("s2.a > 15").select("id")` | `[2]` | no rows |
An array of structs renamed the same way (`ARR2.x`) and `count(s2.b)` (80
against 0 over 100 rows) diverge the same way. On Spark 3.5.9 and 4.1.3 both
engines return NULL for the pruned reads.
### Expected behavior
On Spark 3.4, the native scan resolves the pruned root by its field id as
Spark does. One option is to take the root field ids from `requiredSchema` when
serializing the native data schema.
### Additional context
Found in a nested-type audit of the native scans. Needs the opt-in
`spark.sql.parquet.fieldId.read.enabled`, a read schema that renames a root
struct, and nested schema pruning (on by default). The Spark 3.5+ behavior,
where Spark's own reader loses the root id under pruning, looks like a Spark
regression and is not a Comet problem.
--
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]
