Repository navigation
Native Parquet scan returns NULL for a root column whose file column an unread field-id alias claims #7206
Description
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:
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 330b29c:
| 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 #7196 (comment). 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.
Activity