andygrove opened a new issue, #7206:
URL: https://github.com/apache/datafusion-comet/issues/7206
### Describe the bug
With `spark.sql.parquet.fieldId.read.enabled=true`, the native Parquet scan
returns NULL for a root column that matches a file column by name when an
unread field of the read schema selects the same file column by id. Spark reads
the value.
Take a file with `A` (id 1) = 7 and `z` (id 2) = 9, read with the schema `[x
(id 1), a (no id), z (id 2)]`. The query reads `a` and `z` but not `x`. Spark's
`clipParquetGroupFields` only resolves the required schema `[a, z]`, so `a`
matches `A` by name and both Spark readers return 7.
Comet resolves the root over the whole data schema instead.
`CometNativeScan` serializes the relation's data schema together with a
projection, and `SparkPhysicalExprAdapterFactory::create` receives that data
schema as its logical file schema, the same mechanism behind #7006. The unread
`x` claims `A` by id, which outranks `a`'s name claim, so
`remap_physical_schema` renames `A` to `x` and `a` reads as a missing column.
Id matching is on only when the required schema carries an id, which is why
`select("a")` alone reads correctly.
### Steps to reproduce
In a `CometTestBase` suite on the default Spark 4.1.3 profile:
```scala
def withId(id: Int) =
new MetadataBuilder().putLong(ParquetUtils.FIELD_ID_METADATA_KEY,
id.toLong).build()
val writeSchema = new StructType()
.add("A", LongType, true, withId(1))
.add("z", LongType, true, withId(2))
withSQLConf(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true") {
spark
.createDataFrame(spark.sparkContext.parallelize(Seq(Row(7L, 9L))),
writeSchema)
.write
.parquet(path)
}
val readSchema = new StructType()
.add("x", LongType, true, withId(1))
.add("a", LongType, true)
.add("z", LongType, true, withId(2))
withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
spark.read.schema(readSchema).parquet(path).select("a", "z").collect()
}
```
On `main` at 330b29cf10:
| Query | Spark vectorized | Spark parquet-mr | Comet native scan |
|---|---|---|---|
| `select("a", "z")` | `[7, 9]` | `[7, 9]` | `[null, 9]` |
| `select("a", "z")`, schema order `[a, x, z]` | `[7, 9]` | `[7, 9]` |
`[null, 9]` |
| `select("a", "z")`, case-sensitive, file column named `a` | `[7, 9]` |
`[7, 9]` | `[null, 9]` |
| `select("a")` | `[7]` | `[7]` | `[7]` |
| `select("x", "a", "z")` | `[7, <stale memory>, 9]` | `[7, null, 9]` | `[7,
null, 9]` |
### Expected behavior
Comet should return `[7, 9]`, as both Spark readers do. Root claims should
be resolved among the columns the read requires, the way Spark clips only the
required schema.
### Additional context
Found while reviewing #7196. That PR adds a JVM fallback for the repeated-id
version of this shape, a required root that shares its id with another root of
the data schema. This shape repeats no id, so the fallback doesn't cover it.
@sam-1112 suggested limiting claim arbitration to the required roots in
https://github.com/apache/datafusion-comet/pull/7196#discussion_r4369441283.
That would fix both shapes, and the fallback could then go. `branch-1.2` has
the same root resolution code as `main`.
The shape needs field-id reads and a read schema that mixes fields with and
without ids, so it should be rare in practice.
--
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]
