[
https://issues.apache.org/jira/browse/SPARK-60262?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60262:
-----------------------------------
Labels: pull-request-available (was: )
> Parquet field-id reads return zeros or uninitialized memory when an id match
> and a name match select the same column
> --------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60262
> URL: https://issues.apache.org/jira/browse/SPARK-60262
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Andrew Grove
> Priority: Major
> Labels: pull-request-available
>
> _Drafted with LLM assistance (Claude Code); verified by running the
> reproduction below on Spark 4.1.3, and by reading the same code in Spark
> 4.2.0._
> h3. Problem
> With {{spark.sql.parquet.fieldId.read.enabled=true}},
> {{ParquetReadSupport.clipParquetGroupFields}} resolves each requested field
> on its own. A field that carries a field id resolves by id
> ({{matchIdField}}), and any other field resolves by name. Two requested
> fields can therefore select the same Parquet column. 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. Primitive types are not
> clipped, so the requested Parquet schema lists {{A}} twice.
> Neither reader returns {{A}}'s values for both fields, and the vectorized
> reader returns wrong values without an error:
> * *Vectorized reader.* {{VectorizedParquetRecordReader}} builds one
> {{VectorizedColumnReader}} per requested leaf. Both leaves have the same
> {{ColumnDescriptor}}. parquet-java's
> {{ColumnChunkPageReadStore.getPageReader}} looks up one
> {{ColumnChunkPageReader}} per descriptor, and its {{readPage}} polls a queue.
> The first reader drains the pages. For the second, {{readPage()}} gets no
> page and returns {{-1}}, and {{readBatch}} stops as if it had read every
> page. Its vector is never written, and the null mask isn't either, so the
> rows read as non-null:
> ** With on-heap vectors, the default, they read {{0}}.
> ** With off-heap vectors they read uninitialized memory. Off-heap vectors are
> controlled by {{spark.sql.columnVector.offheap.enabled}}, which follows
> {{spark.memory.offHeap.enabled}} from Spark 4.0 on.
> ** Whichever field comes first in the requested schema gets the values.
> * *parquet-mr reader* ({{spark.sql.parquet.enableVectorizedReader=false}}).
> {{ParquetRowConverter}} maps the clipped Parquet fields back to requested
> fields by id first. Both copies of {{A}} carry id 1, so both map to {{x}},
> and {{a}} reads null.
> h3. Reproduce
> {code:scala}
> import org.apache.spark.sql.Row
> import org.apache.spark.sql.types._
> 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()
> {code}
> ||Requested fields||Vectorized, on-heap vectors (default)||Vectorized,
> off-heap vectors||parquet-mr||
> |{{x}} (id 1), {{a}}|{{x}} = 1, 2, 3; {{a}} = 0, 0, 0|{{x}} = 1, 2, 3; {{a}}
> = 0, 0, 1634492735|{{x}} = 1, 2, 3; {{a}} = null|
> |{{a}}, {{x}} (id 1)|{{a}} = 1, 2, 3; {{x}} = 0, 0, 0|{{a}} = 1, 2, 3; {{x}}
> = stale memory|{{a}} = null; {{x}} = 1, 2, 3|
> |struct {{s}} holding {{x}} (id 1), {{a}}|{{a}} = 0, 0, 0|{{a}} = stale
> memory|{{a}} = null|
> For the struct case, write {{A}} inside a struct {{s}} and request {{s}} with
> the fields {{x}} and {{a}}. The off-heap values change between runs
> ({{1634492735}}, {{489223743}}, {{1403037696}}, ...). These results are from
> Spark 4.1.3, where nested column vectorization is on by default.
> h3. Expected behavior
> Both fields read {{1, 2, 3}}, because {{clipParquetGroupFields}} resolves
> each of them to {{A}}. Failing the read would also be reasonable. Returning
> zeros or uninitialized memory without an error is not.
> h3. Possible fix
> * *Guard.* Make {{VectorizedColumnReader.readBatch}} fail, instead of
> returning, when it runs out of pages with rows still to read in a column that
> isn't repeated. That would turn the wrong values into an error, whatever else
> changes.
> * *Vectorized reader.* In {{VectorizedParquetRecordReader}}, read a column
> once when several requested leaves resolve to it, and fill each of their
> vectors from that read.
> * *parquet-mr reader.* {{clipParquetGroupFields}} returns one clipped field
> per requested field, in requested order. So {{ParquetRowConverter}} could
> pair them by position instead of by id or name. Someone who knows why it
> matches by id would need to confirm that is safe.
> h3. Context
> I found this while comparing DataFusion Comet's native Parquet scan with
> Spark for this shape ([apache/datafusion-comet issue
> 6994|https://github.com/apache/datafusion-comet/issues/6994]). Field-id reads
> arrived in Spark 3.3 ([apache/spark PR
> 35385|https://github.com/apache/spark/pull/35385], SPARK-38094). The shape
> needs a requested struct that mixes fields with and without field ids, so it
> is uncommon: Iceberg and Delta give every column an id.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
