[
https://issues.apache.org/jira/browse/SPARK-60293?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Grove updated SPARK-60293:
---------------------------------
Labels: correctness pull-request-available (was: pull-request-available)
> 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
> Priority: Major
> Labels: correctness, pull-request-available
>
> 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]
