andygrove opened a new pull request, #59552: URL: https://github.com/apache/spark/pull/59552
### What changes were proposed in this pull request? With `spark.sql.parquet.fieldId.read.enabled=true`, `ParquetReadSupport.clipParquetGroupFields` matches each requested field that has a field id to the Parquet field with that id. Two later steps still went by name, so a read schema that gives a requested field the name of a different field in the file read wrong. Two fields that swap names but keep their ids are the simplest case. This PR makes both steps resolve by field id, like the clipping does. - **Filter pushdown.** `ParquetFilters` resolved a pushed column by its name in the file's Parquet schema. It now takes an optional `fieldIdReadSchema`. When it is set, each pushed column name is looked up in that schema and resolved to the Parquet column the same way the clipping selects it: by field id if the field has one, otherwise by name. A field with an id that the file does not have is not pushed down, instead of being pushed down against a file column that has the same name. The DSv1 scan (`ParquetFileFormat`) and the DSv2 reader factory (`ParquetPartitionReaderFactory`) pass their `dataSchema` when field-id reads are enabled and the schema has field ids. Reads without field ids resolve as before. The DSv2 planning-time check in `ParquetScanBuilder` is unchanged, because its Parquet schema is derived from the read schema, so names and ids agree there. - **Vectorized reader.** `ParquetToSparkSchemaConverter.convertInternal` took each clipped Parquet field's Spark read type from the requested field with the same name, and used the field id only when no requested field had that name. It now prefers the requested field with the Parquet field's id when `useFieldId` is set, and falls back to the name. ### Why are the changes needed? Take a file with `s<a int id 21, b string id 22>` and the read schema `s<b int id 21, a string id 22>`, so `s.b` is the file's `s.a`: - `WHERE s.b > 15`, `WHERE s.a = 'y'` and even `WHERE s.b IS NOT NULL` returned no rows, with no error, in both readers. `ParquetFilters` evaluated the filter against the file's `s.b` statistics, and the `IsNotNull` it pushes names a file column that the id-clipped read does not project, so parquet-mr's column index filter drops every page. With two `int` fields, `SELECT *` was right, but `WHERE s.b > 15` still skipped the row group on the other column's statistics. - `SELECT *` failed in the vectorized reader with `SchemaColumnConvertNotSupportedException: column: [s, a], physicalType: INT32, logicalType: string`, because the clipped column `s.a` (INT32) got the read type of the requested `a` (string). The same happened for a swap at the top level. The parquet-mr reader read those cases correctly. `spark.sql.parquet.filterPushdown=false` returned the right rows, and so did Comet's native scan, which resolves these fields by id. The shape needs field-id reads and a read schema that reuses a file field's name for another field. Table formats with field ids can produce it through renames. SPARK-52267 added the id fallback in `convertInternal` and kept name-first matching to keep that change small. ### Does this PR introduce _any_ user-facing change? Yes, for a field-id read where a requested field has the name of a different field in the file. Filters on such a field returned no rows or the wrong rows, and now return the rows that match. The vectorized reader failed on swapped names with different types, and now reads them. Reads that do not reuse a file field's name for another field are not affected, and neither are reads without field ids or with `spark.sql.parquet.fieldId.read.enabled=false`. ### How was this patch tested? New tests in `ParquetFieldIdIOSuite`, each run on the DSv1 and the DSv2 source with both the vectorized and the parquet-mr reader: - swapped names with different types: `SELECT *`, `s.b > 15`, `s.a = 'y'` and `IS NOT NULL` on either field; - swapped names with the same type: `s.b > 15`, `s.b < 15`, `s.a = 2`, `s.a > 15` and a conjunction, which depend on the row group statistics of the right column; - swapped names at the top level; - a requested field whose id the file does not have, which reads as null and is not pushed down against the file's field of that name (`a IS NULL` returns every row); - a unit test of `ParquetFilters` with and without `fieldIdReadSchema`, including nested fields and a name that the file does not have. The first four fail without the change (wrong rows, and `PARQUET_COLUMN_DATA_TYPE_MISMATCH` for the top-level swap in the vectorized reader) and pass with it. Also ran, all passing, on this change applied to a master from a few commits earlier (the tests were not rerun on the exact base of this PR in a local run; the merged tree was compiled, and the CI run on this PR covers the tests): `ParquetFieldIdIOSuite`, `ParquetFieldIdSchemaSuite`, `ParquetSchemaSuite`, `ParquetV1FilterSuite`, `ParquetV2FilterSuite`, `ParquetIOSuite`, `ParquetV1SchemaPruningSuite`, `ParquetV2SchemaPruningSuite` (988 tests, 18 suites), and `ParquetV1QuerySuite`, `ParquetV2QuerySuite`, `ParquetFileFormatV1Suite`, `ParquetFileFormatV2Suite` (128 tests). ### 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]
