You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2021/09/22 09:42:17 UTC
[spark] branch branch-3.2 updated: [SPARK-36803][SQL] Fix ArrayType
conversion when reading Parquet files written in legacy mode
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new fc0b85f [SPARK-36803][SQL] Fix ArrayType conversion when reading Parquet files written in legacy mode
fc0b85f is described below
commit fc0b85fb2652b8c9c856a0aa051f91b985dacfd6
Author: Ivan Sadikov <iv...@databricks.com>
AuthorDate: Wed Sep 22 17:40:39 2021 +0800
[SPARK-36803][SQL] Fix ArrayType conversion when reading Parquet files written in legacy mode
### What changes were proposed in this pull request?
This PR fixes an issue when reading of a Parquet file written with legacy mode would fail due to incorrect Parquet LIST to ArrayType conversion.
The issue arises when using schema evolution and utilising the parquet-mr reader. 2-level LIST annotated types could be parsed incorrectly as 3-level LIST annotated types because their underlying element type does not match the full inferred Catalyst schema.
### Why are the changes needed?
It appears to be a long-standing issue with the legacy mode due to the imprecise check in ParquetRowConverter that was trying to determine Parquet backward compatibility using Catalyst schema: `DataType.equalsIgnoreCompatibleNullability(guessedElementType, elementType)` in https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala#L606.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added a new test case in ParquetInteroperabilitySuite.scala.
Closes #34044 from sadikovi/parquet-legacy-write-mode-list-issue.
Authored-by: Ivan Sadikov <iv...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit ec26d94eacb22562b8f5d60d12cf153d8ef3fd50)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../datasources/parquet/ParquetRowConverter.scala | 16 +++++--
.../parquet/ParquetSchemaConverter.scala | 2 +-
.../parquet/ParquetInteroperabilitySuite.scala | 53 ++++++++++++++++++++++
3 files changed, 66 insertions(+), 5 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
index e0af5d8..583b4ba 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
@@ -574,7 +574,7 @@ private[parquet] class ParquetRowConverter(
val repeatedType = parquetSchema.getType(0)
val elementType = catalystSchema.elementType
- // At this stage, we're not sure whether the repeated field maps to the element type or is
+ // At this stage, we need to figure out if the repeated field maps to the element type or is
// just the syntactic repeated group of the 3-level standard LIST layout. Take the following
// Parquet LIST-annotated group type as an example:
//
@@ -598,12 +598,20 @@ private[parquet] class ParquetRowConverter(
//
// ARRAY<STRUCT<element: STRUCT<element: INT>>>
//
+ //
// Here we try to convert field `list` into a Catalyst type to see whether the converted type
- // matches the Catalyst array element type. If it doesn't match, then it's case 1; otherwise,
- // it's case 2.
+ // matches the Catalyst array element type.
+ //
+ // If the guessed element type from the above does not match the Catalyst type (for example,
+ // in case of schema evolution), we need to check if the repeated type matches one of the
+ // backward-compatibility rules for legacy LIST types (see the link above).
+ //
+ // If the element type does not match the Catalyst type and the underlying repeated type
+ // does not belong to the legacy LIST type, then it is case 1; otherwise, it is case 2.
val guessedElementType = schemaConverter.convertField(repeatedType)
+ val isLegacy = schemaConverter.isElementType(repeatedType, parquetSchema.getName)
- if (DataType.equalsIgnoreCompatibleNullability(guessedElementType, elementType)) {
+ if (DataType.equalsIgnoreCompatibleNullability(guessedElementType, elementType) || isLegacy) {
// If the repeated field corresponds to the element type, creates a new converter using the
// type of the repeated field.
newConverter(repeatedType, elementType, new ParentContainerUpdater {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
index f3bfd99..f3ecd79 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala
@@ -266,7 +266,7 @@ class ParquetToSparkSchemaConverter(
// Here we implement Parquet LIST backwards-compatibility rules.
// See: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules
// scalastyle:on
- private def isElementType(repeatedType: Type, parentName: String): Boolean = {
+ private[parquet] def isElementType(repeatedType: Type, parentName: String): Boolean = {
{
// For legacy 2-level list types with primitive element type, e.g.:
//
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
index b1d930a..a7395a6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
@@ -29,6 +29,7 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{ArrayType, IntegerType, StructField, StructType}
class ParquetInteroperabilitySuite extends ParquetCompatibilityTest with SharedSparkSession {
test("parquet files with different physical schemas but share the same logical schema") {
@@ -96,6 +97,58 @@ class ParquetInteroperabilitySuite extends ParquetCompatibilityTest with SharedS
}
}
+ test("SPARK-36803: parquet files with legacy mode and schema evolution") {
+ // This test case writes arrays in Parquet legacy mode and schema evolution and verifies that
+ // the data can be correctly read back.
+
+ Seq(false, true).foreach { legacyMode =>
+ withSQLConf(SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key -> legacyMode.toString) {
+ withTempPath { tableDir =>
+ val schema1 = StructType(
+ StructField("col-0", ArrayType(
+ StructType(
+ StructField("col-0", IntegerType, true) ::
+ Nil
+ ),
+ containsNull = false // allows to create 2-level Parquet LIST type in legacy mode
+ )) ::
+ Nil
+ )
+ val row1 = Row(Seq(Row(1)))
+ val df1 = spark.createDataFrame(spark.sparkContext.parallelize(row1 :: Nil, 1), schema1)
+ df1.write.parquet(tableDir.getAbsolutePath)
+
+ val schema2 = StructType(
+ StructField("col-0", ArrayType(
+ StructType(
+ StructField("col-0", IntegerType, true) ::
+ StructField("col-1", IntegerType, true) :: // additional field
+ Nil
+ ),
+ containsNull = false
+ )) ::
+ Nil
+ )
+ val row2 = Row(Seq(Row(1, 2)))
+ val df2 = spark.createDataFrame(spark.sparkContext.parallelize(row2 :: Nil, 1), schema2)
+ df2.write.mode("append").parquet(tableDir.getAbsolutePath)
+
+ // Reading of data should succeed and should not fail with
+ // java.lang.ClassCastException: optional int32 col-0 is not a group
+ withAllParquetReaders {
+ checkAnswer(
+ spark.read.schema(schema2).parquet(tableDir.getAbsolutePath),
+ Seq(
+ Row(Seq(Row(1, null))),
+ Row(Seq(Row(1, 2)))
+ )
+ )
+ }
+ }
+ }
+ }
+ }
+
test("parquet timestamp conversion") {
// Make a table with one parquet file written by impala, and one parquet file written by spark.
// We should only adjust the timestamps in the impala file, and only if the conf is set
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org