peter-toth commented on code in PR #59997:
URL: https://github.com/apache/spark/pull/59997#discussion_r4443495194
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala:
##########
@@ -70,6 +194,24 @@ class ParquetV2SchemaPruningSuite extends
ParquetSchemaPruningSuite {
.sparkConf
.set(SQLConf.USE_V1_SOURCE_LIST, "")
+ // V1 uses the session setting for both clipping and conversion. Only V2
honors the scan option.
+ test("SPARK-60788: row reader uses the scan's field ID setting") {
+ withTempPath { dir =>
+ def fieldId(id: Long): Metadata =
+ new MetadataBuilder().putLong(ParquetUtils.FIELD_ID_METADATA_KEY,
id).build()
+ val path = dir.getCanonicalPath
+ spark.range(1).select(col("id").as("x", fieldId(1))).write.parquet(path)
+ val schema = new StructType().add("x", LongType, nullable = true,
metadata = fieldId(2))
+ val key = SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key
+ for (useIds <- Seq(true, false)) {
+ withSQLConf(vectorizedReaderEnabledKey -> "false", key ->
(!useIds).toString) {
+ val data = spark.read.option(key,
useIds).schema(schema).parquet(path)
+ checkAnswer(data, if (useIds) Row(null) else Row(0L))
Review Comment:
**Finding 5.** This test passes on base. Its read schema asks for ID 2,
which the file doesn't have. So both arms give the same answer on base as here.
The bug that reading `useFieldId` from the converter fixes needs an ID that
does match. I wrote a file column `x` with ID 1. Then I read `x` with ID 2 and
`y` with ID 1 in V2, on base:
- option `true`, session `false`: row reader `[0, null]`, vectorized `[null,
0]`
- option `false`, session `true`: row reader `[null, 0]`, vectorized `[0,
null]`
So the base row reader fills the wrong column in both arms. This PR returns
the vectorized answer in both. With the change below, the test fails on base
(`[0,null]` instead of `[null,0]`) and passes here:
```suggestion
// The file column matches x by name and y by ID.
val schema = new StructType()
.add("x", LongType, nullable = true, metadata = fieldId(2))
.add("y", LongType, nullable = true, metadata = fieldId(1))
val key = SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key
for (useIds <- Seq(true, false)) {
withSQLConf(vectorizedReaderEnabledKey -> "false", key ->
(!useIds).toString) {
val data = spark.read.option(key,
useIds).schema(schema).parquet(path)
checkAnswer(data, if (useIds) Row(null, 0L) else Row(0L, null))
```
--
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]
