You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2021/04/09 18:54:27 UTC
[spark] branch branch-3.1 updated: [SPARK-34963][SQL] Fix nested
column pruning for extracting case-insensitive struct field from array of
struct
This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new abc9a4f [SPARK-34963][SQL] Fix nested column pruning for extracting case-insensitive struct field from array of struct
abc9a4f is described below
commit abc9a4f8e7c3ebc37e6efc299b2d95443d3a2565
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Fri Apr 9 11:52:55 2021 -0700
[SPARK-34963][SQL] Fix nested column pruning for extracting case-insensitive struct field from array of struct
### What changes were proposed in this pull request?
This patch proposes a fix of nested column pruning for extracting case-insensitive struct field from array of struct.
### Why are the changes needed?
Under case-insensitive mode, nested column pruning rule cannot correctly push down extractor of a struct field of an array of struct, e.g.,
```scala
val query = spark.table("contacts").select("friends.First", "friends.MiDDle")
```
Error stack:
```
[info] java.lang.IllegalArgumentException: Field "First" does not exist.
[info] Available fields:
[info] at org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:274)
[info] at org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:274)
[info] at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
[info] at scala.collection.AbstractMap.getOrElse(Map.scala:59)
[info] at org.apache.spark.sql.types.StructType.apply(StructType.scala:273)
[info] at org.apache.spark.sql.execution.ProjectionOverSchema$$anonfun$getProjection$3.apply(ProjectionOverSchema.scala:44)
[info] at org.apache.spark.sql.execution.ProjectionOverSchema$$anonfun$getProjection$3.apply(ProjectionOverSchema.scala:41)
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Unit test
Closes #32059 from viirya/fix-array-nested-pruning.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
(cherry picked from commit 364d1eaf10f51c357f507325557fb076140ced2c)
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../expressions/ProjectionOverSchema.scala | 9 +++--
.../sql/catalyst/expressions/SelectedField.scala | 6 +++-
.../execution/datasources/SchemaPruningSuite.scala | 42 ++++++++++++++++++++++
3 files changed, 54 insertions(+), 3 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ProjectionOverSchema.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ProjectionOverSchema.scala
index 241c761..03b5517 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ProjectionOverSchema.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ProjectionOverSchema.scala
@@ -41,9 +41,14 @@ case class ProjectionOverSchema(schema: StructType) {
case a: GetArrayStructFields =>
getProjection(a.child).map(p => (p, p.dataType)).map {
case (projection, ArrayType(projSchema @ StructType(_), _)) =>
+ // For case-sensitivity aware field resolution, we should take `ordinal` which
+ // points to correct struct field.
+ val selectedField = a.child.dataType.asInstanceOf[ArrayType]
+ .elementType.asInstanceOf[StructType](a.ordinal)
+ val prunedField = projSchema(selectedField.name)
GetArrayStructFields(projection,
- projSchema(a.field.name),
- projSchema.fieldIndex(a.field.name),
+ prunedField.copy(name = a.field.name),
+ projSchema.fieldIndex(selectedField.name),
projSchema.size,
a.containsNull)
case (_, projSchema) =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SelectedField.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SelectedField.scala
index f2acb75..39dfdf9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SelectedField.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SelectedField.scala
@@ -75,7 +75,11 @@ object SelectedField {
val field = c.childSchema(c.ordinal)
val newField = field.copy(dataType = dataTypeOpt.getOrElse(field.dataType))
selectField(c.child, Option(struct(newField)))
- case GetArrayStructFields(child, field, _, _, containsNull) =>
+ case GetArrayStructFields(child, _, ordinal, _, containsNull) =>
+ // For case-sensitivity aware field resolution, we should take `ordinal` which
+ // points to correct struct field.
+ val field = child.dataType.asInstanceOf[ArrayType]
+ .elementType.asInstanceOf[StructType](ordinal)
val newFieldDataType = dataTypeOpt match {
case None =>
// GetArrayStructFields is the top level extractor. This means its result is
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala
index c907321..765d2fc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala
@@ -774,4 +774,46 @@ abstract class SchemaPruningSuite
assert(scanSchema === expectedScanSchema)
}
}
+
+ testSchemaPruning("SPARK-34963: extract case-insensitive struct field from array") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val query1 = spark.table("contacts")
+ .select("friends.First", "friends.MiDDle")
+ checkScan(query1, "struct<friends:array<struct<first:string,middle:string>>>")
+ checkAnswer(query1,
+ Row(Array.empty[String], Array.empty[String]) ::
+ Row(Array("Susan"), Array("Z.")) ::
+ Row(null, null) ::
+ Row(null, null) :: Nil)
+
+ val query2 = spark.table("contacts")
+ .where("friends.First is not null")
+ .select("friends.First", "friends.MiDDle")
+ checkScan(query2, "struct<friends:array<struct<first:string,middle:string>>>")
+ checkAnswer(query2,
+ Row(Array.empty[String], Array.empty[String]) ::
+ Row(Array("Susan"), Array("Z.")) :: Nil)
+ }
+ }
+
+ testSchemaPruning("SPARK-34963: extract case-insensitive struct field from struct") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val query1 = spark.table("contacts")
+ .select("Name.First", "NAME.MiDDle")
+ checkScan(query1, "struct<name:struct<first:string,middle:string>>")
+ checkAnswer(query1,
+ Row("Jane", "X.") ::
+ Row("Janet", null) ::
+ Row("Jim", null) ::
+ Row("John", "Y.") :: Nil)
+
+ val query2 = spark.table("contacts")
+ .where("Name.MIDDLE is not null")
+ .select("Name.First", "NAME.MiDDle")
+ checkScan(query2, "struct<name:struct<first:string,middle:string>>")
+ checkAnswer(query2,
+ Row("Jane", "X.") ::
+ Row("John", "Y.") :: Nil)
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org