Andrew Grove created SPARK-60293:
------------------------------------
Summary: Parquet field-id reads: pushed filters and vectorized
read types resolve swapped fields by name, not by id
Key: SPARK-60293
URL: https://issues.apache.org/jira/browse/SPARK-60293
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Andrew Grove
h3. Problem
With {{spark.sql.parquet.fieldId.read.enabled=true}},
{{ParquetReadSupport.clipParquetGroupFields}} matches each requested field to
the file field with the same field id. Two later steps still go by name, so a
read schema in which a requested field's name belongs to a different field in
the file reads wrong. The simplest case is two fields that swap names but keep
their ids, which a rename in a table format with field ids can produce.
* *Pushed filters return no rows, without an error.* {{ParquetFilters}}
resolves a pushed column by its name in the file's Parquet schema
({{nameToParquetField}}), not by id. A filter on the requested {{s.b}} (id 21,
the file's {{s.a}}) is evaluated against the file's {{s.b}}. Both readers are
affected, and {{spark.sql.parquet.filterPushdown=false}} gives the right rows.
** A value predicate prunes row groups with the other column's statistics. With
the file's {{s.a}} = 10, 20 and {{s.b}} = 1, 2, {{WHERE s.b > 15}} means the
file's {{s.a > 15}}, but it is checked against the file's {{s.b}}, whose max is
2, so the row group is skipped and no rows come back.
** Spark also pushes {{IsNotNull(s.b)}} for a filter on {{s.b}}. That names a
file column the id-clipped read does not project, and parquet-mr's
{{ColumnIndexFilter}} treats a predicate column outside the projection as all
NULL, so it drops every page. When the swapped fields have different types,
that alone empties {{WHERE s.b > 15}}, {{WHERE s.a = 'y'}} and even {{WHERE s.b
IS NOT NULL}}. Setting {{parquet.filter.columnindex.enabled=false}} restores
those rows. Disabling the stats, dictionary or bloom filters does not.
* *The vectorized reader fails when the swapped fields have different types.*
{{ParquetToSparkSchemaConverter.convertInternal}} takes each clipped Parquet
field's Spark read type from the requested field with the same name, and uses
the field id only when no requested field has that name. The clipped column
{{s.a}} (INT32, id 21) gets the read type of the requested {{a}} (string, id
22), so the read fails with {{SchemaColumnConvertNotSupportedException: column:
[s, a], physicalType: INT32, logicalType: string}}. The same happens for a swap
at the root. Swapped fields of the same type read the right values, and the
parquet-mr reader reads both cases correctly.
h3. Reproduce
{code:scala}
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
def id(n: Int) = new MetadataBuilder().putLong("parquet.field.id",
n.toLong).build()
val fileSchema = StructType(Seq(
StructField("id", IntegerType, nullable = true, id(1)),
StructField("s", StructType(Seq(
StructField("a", IntegerType, nullable = true, id(21)),
StructField("b", StringType, nullable = true, id(22)))), nullable = true,
id(2))))
spark
.createDataFrame(
spark.sparkContext.parallelize(Seq(Row(1, Row(10, "x")), Row(2, Row(20,
"y")), Row(3, null)), 1),
fileSchema)
.write
.parquet("/tmp/field-id-swap")
spark.conf.set("spark.sql.parquet.fieldId.read.enabled", "true")
// Same ids, names swapped: b (id 21) is the file's a, and a (id 22) is the
file's b.
val readSchema = StructType(Seq(
StructField("id", IntegerType, nullable = true, id(1)),
StructField("s", StructType(Seq(
StructField("b", IntegerType, nullable = true, id(21)),
StructField("a", StringType, nullable = true, id(22)))), nullable = true,
id(2))))
val df = spark.read.schema(readSchema).parquet("/tmp/field-id-swap")
df.where("s.b > 15").select("id").show()
df.show()
{code}
Results on Spark 4.1.3:
||Query||Vectorized||parquet-mr||Either reader,
{{parquet.filter.columnindex.enabled=false}}||Either reader,
{{spark.sql.parquet.filterPushdown=false}}||
|{{WHERE s.b > 15}}|no rows|no rows|{{[2]}}|{{[2]}}|
|{{WHERE s.a = 'y'}}|no rows|no rows|{{[2]}}|{{[2]}}|
|{{WHERE s.b IS NOT NULL}}|no rows|no rows|{{[1] [2]}}|not run|
{{SELECT *}} fails in the vectorized reader with
{{SchemaColumnConvertNotSupportedException: column: [s, a], physicalType:
INT32, logicalType: string}}, with pushdown on or off, while parquet-mr returns
{{[1,[10,x]] [2,[20,y]] [3,null]}}. Reading one of the swapped fields alone, as
in {{SELECT id, s.b}}, gives the right values in both readers.
With two {{int}} fields (the file's {{s.a}} = 10, 20 and {{s.b}} = 1, 2, read
as {{s<b int id 21, a int id 22>}}), {{SELECT *}} reads the right values, but
{{WHERE s.b > 15}} still returns no rows in both readers, even with
{{parquet.filter.columnindex.enabled=false}}. It returns {{[2]}} with
{{spark.sql.parquet.filterPushdown=false}}.
h3. Expected behavior
With field-id reads, pushed filters and read types resolve requested fields by
id, as {{clipParquetGroupFields}} does. {{WHERE s.b > 15}} returns {{2}}, and
{{SELECT *}} returns {{[1,[10,x]] [2,[20,y]] [3,null]}} in both readers.
h3. Possible fix
* *Filter pushdown.* Resolve pushed columns through the id-clipped schema, so
that each pushed column maps to the file path of the field its id selects. A
pushed column whose id the file does not have, or that matches ambiguously, is
not pushed down.
* *Vectorized reader.* In {{convertInternal}}, prefer the field id when
{{useFieldId}} is set and the Parquet field carries one, as
{{clipParquetGroupFields}} does. SPARK-52267 (apache/spark PR 50990) added the
id fallback and kept name-first matching on purpose, to keep the change small.
h3. Context
Found while comparing DataFusion Comet's native Parquet scan with Spark
(apache/datafusion-comet issue 7040). Comet resolves these fields by id and
returns the expected rows. SPARK-60262 is a related field-id read problem,
where an id match and a name match select the same column. The shape needs
field-id reads and a read schema that reuses a file field's name for another
field, so it is uncommon.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
