andygrove opened a new issue, #7006:
URL: https://github.com/apache/datafusion-comet/issues/7006
### Describe the bug
With `spark.sql.parquet.fieldId.read.enabled=true`, the native Parquet scan
fails on a file in which two root columns carry the same field id whenever the
table schema has a field with that id, even when the query never reads that
field. Spark reads the file. `ParquetReadSupport.clipParquetGroupFields` only
resolves the required schema, and `matchIdField` raises
`_LEGACY_ERROR_TEMP_2094` only for a required field whose id matches more than
one file field.
Comet runs the check over the whole data schema instead. `CometNativeScan`
serializes the relation's data schema (`nativeDataSchema` drops only variant
columns the read does not request) together with a projection.
`init_datasource_exec` builds the `TableSchema` from that data schema, and
DataFusion's Parquet opener passes `table_schema.file_schema()` to
`SparkPhysicalExprAdapterFactory::create`. `remap_physical_schema` then raises
the duplicate id error while it resolves every logical root field. The
ambiguous name check already waits for a column the read references
(`check_duplicate_columns`), so only the id check fires early.
### 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("p", IntegerType, true, withId(7))
.add("q", IntegerType, true, withId(7))
.add("b", IntegerType, true, withId(3))
spark
.createDataFrame(spark.sparkContext.parallelize(Seq(Row(1, 2, 3), Row(4,
5, 6))), writeSchema)
.write
.parquet(path)
val readSchema = new StructType()
.add("u", IntegerType, true, withId(7))
.add("b", IntegerType, true, withId(3))
withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
spark.read.schema(readSchema).parquet(path).select("b").collect()
}
```
On `main` at 66c63bd307:
| Query | Spark | Comet native scan |
|---|---|---|
| `select("b")` | `3`, `6` | `Found duplicate field(s) "7": [p, q] in id
mapping mode.` |
| `where("b > 3").select("b")` | `6` | the same error |
| `select("u")` | the same error | the same error |
| the whole read | the same error | the same error |
### Expected behavior
Comet should read `b` as Spark does, and raise the duplicate id error only
when the read references a column whose id matches more than one file field,
the way `check_duplicate_columns` already defers an ambiguous name.
### Additional context
@comphead found this while reviewing #6990:
https://github.com/apache/datafusion-comet/pull/6990#discussion_r4277806405.
`branch-1.1` has the same check, and #6990 keeps the behavior. On top of #6990,
one way to fix it is for `FieldResolver::select` to return every id match, as
it already does for a name. `resolve` would then raise the error for a nested
field, and the root would record the ambiguity for `check_duplicate_columns` to
raise when the read references the column.
The shape needs a file that repeats a field id, 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]
