andygrove opened a new pull request, #59553: URL: https://github.com/apache/spark/pull/59553
### What changes were proposed in this pull request? Backport of #59510 to branch-4.3, without the migration guide entry, as requested in https://github.com/apache/spark/pull/59510#issuecomment-6219245530. The cherry-pick applies as is, so the rest of the change is identical to #59510. With `spark.sql.parquet.fieldId.read.enabled=true`, `ParquetReadSupport.clipParquetGroupFields` resolves each requested field on its own: a field with a field id by id, and any other field by name. Two requested fields can therefore be clipped to the same Parquet field, and neither reader can read a column twice (see below). This PR makes such a read fail with a clear error instead of returning wrong values: - `clipParquetGroupFields` throws the new `PARQUET_FIELD_REQUESTED_MORE_THAN_ONCE` error when two or more requested fields of a struct are clipped to the same Parquet field. That covers a field id match plus a name match, and two fields with the same id, at the top level and inside nested structs. The check is in the clipping code that both readers use, so it covers the vectorized reader and the parquet-mr reader. - `VectorizedColumnReader.readBatch` throws a `ParquetDecodingException` when a column without repeated values runs out of pages with rows still to read, instead of returning. With the check above nothing in Spark reaches this, but returning leaves the rest of the vector unwritten without any signal, so this keeps a column that is ever read twice, or a truncated column chunk, from quietly returning wrong values. Repeated columns still stop at the end of the pages, because the end of a list is not known until then. ### Why are the changes needed? Take a file column `A` with field id 1. A requested `x` with id 1 selects it by id, and a requested `a` without an id selects it by name, so the requested Parquet schema lists `A` twice. Neither reader returns `A`'s values for both fields, and neither reports an error: - The vectorized reader builds one `VectorizedColumnReader` per requested leaf, but a Parquet page store has a single queue of pages per column. The first reader drains it. For the second, `readPage()` returns -1 and `readBatch` stops as if it had read every page, so its vector is never written, and its rows read as non-null `0` with on-heap vectors, or as whatever the memory holds with off-heap vectors. - The parquet-mr reader (`spark.sql.parquet.enableVectorizedReader=false`) has `ParquetRowConverter` map the clipped Parquet fields back to the requested fields by id first. Both copies of `A` carry id 1, so both map to `x`, and `a` reads null. ```scala val id1 = new MetadataBuilder().putLong("parquet.field.id", 1L).build() spark.createDataFrame( spark.sparkContext.parallelize(Seq(Row(1), Row(2), Row(3)), 1), StructType(Seq(StructField("A", IntegerType, nullable = true, id1)))) .write.parquet("/tmp/field-id-twice") spark.conf.set("spark.sql.parquet.fieldId.read.enabled", "true") val x = StructField("x", IntegerType, nullable = true, id1) val a = StructField("a", IntegerType, nullable = true) spark.read.schema(StructType(Seq(x, a))).parquet("/tmp/field-id-twice").show() ``` Before this change, `x` reads 1, 2, 3 and `a` reads 0, 0, 0 with the vectorized reader (the other way round when `a` comes first), and `a` reads null with the parquet-mr reader. A struct holding `x` and `a` behaves the same. Reading both fields correctly would take changes to how both readers map clipped fields back to the requested ones, and a way for the vectorized reader to serve one column to several leaves. This PR is the smaller step of replacing the silent wrong results with an error. Field-id reads (SPARK-38094) have returned these results since 3.3. ### Does this PR introduce _any_ user-facing change? Yes. Reading a schema in which two requested fields select the same Parquet column, which returned wrong values without an error (see above), now fails with `PARQUET_FIELD_REQUESTED_MORE_THAN_ONCE`. The wrong results come from field-id reads, so released versions since 3.3 have them as well. Reads that do not select a column more than once are not affected. ### How was this patch tested? The tests added in #59510 are included: `ParquetFieldIdSchemaSuite`, `ParquetFieldIdIOSuite` and `ParquetVectorizedSuite`. I did not run them locally for this backport; they run in CI on this PR. On master, #59510 ran the packages `org.apache.spark.sql.execution.datasources.parquet` and `org.apache.spark.sql.execution.datasources.v2.parquet` with `FileBasedDataSourceSuite` (55 suites, 2075 tests passed, 7 ignored), and `SparkThrowableSuite`, all passing. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-sonnet-5-5) -- 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]
