Andrew Grove created SPARK-60262:
------------------------------------
Summary: 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
_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]
