Skip to content

Commit

Permalink
[SPARK-42899][SQL] Fix DataFrame.to(schema) to handle the case where …
Browse files Browse the repository at this point in the history
…there is a non-nullable nested field in a nullable field

Fixes `DataFrame.to(schema)` to handle the case where there is a non-nullable nested field in a nullable field.

`DataFrame.to(schema)` fails when it contains non-nullable nested field in nullable field:

```scala
scala> val df = spark.sql("VALUES (1, STRUCT(1 as i)), (NULL, NULL) as t(a, b)")
df: org.apache.spark.sql.DataFrame = [a: int, b: struct<i: int>]
scala> df.printSchema()
root
 |-- a: integer (nullable = true)
 |-- b: struct (nullable = true)
 |    |-- i: integer (nullable = false)

scala> df.to(df.schema)
org.apache.spark.sql.AnalysisException: [NULLABLE_COLUMN_OR_FIELD] Column or field `b`.`i` is nullable while it's required to be non-nullable.
```

No.

Added the related tests.

Closes #40526 from ueshin/issues/SPARK-42899/to_schema.

Authored-by: Takuya UESHIN <ueshin@databricks.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>
(cherry picked from commit 4052058)
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>
  • Loading branch information
ueshin authored and HyukjinKwon committed Mar 23, 2023
1 parent f73b5c5 commit a1b853c
Show file tree
Hide file tree
Showing 2 changed files with 35 additions and 1 deletion.
Expand Up @@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable
import org.apache.spark.sql.catalyst.catalog.CatalogTable.VIEW_STORING_ANALYZED_PLAN
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, TypedImperativeAggregate}
import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, RangePartitioning, RoundRobinPartitioning, SinglePartition}
import org.apache.spark.sql.catalyst.trees.TreeNodeTag
Expand Down Expand Up @@ -118,7 +119,7 @@ object Project {
case (StructType(fields), expected: StructType) =>
val newFields = reorderFields(
fields.zipWithIndex.map { case (f, index) =>
(f.name, GetStructField(col, index))
(f.name, GetStructField(AssertNotNull(col, columnPath), index))
},
expected.fields,
columnPath,
Expand Down
Expand Up @@ -180,6 +180,17 @@ class DataFrameToSchemaSuite extends QueryTest with SharedSparkSession {
checkAnswer(df, Row(Row(1)))
}

test("struct value: compatible field nullability") {
val innerFields = new StructType().add("i", LongType, nullable = false)
val schema = new StructType().add("a", LongType).add("b", innerFields)
val data = sql("VALUES (1, STRUCT(1 as i)), (NULL, NULL) as t(a, b)")
assert(data.schema.fields(1).nullable)
assert(!data.schema.fields(1).dataType.asInstanceOf[StructType].fields(0).nullable)
val df = data.to(schema)
assert(df.schema == schema)
checkAnswer(df, Seq(Row(1, Row(1)), Row(null, null)))
}

test("negative: incompatible field nullability") {
val innerFields = new StructType().add("i", IntegerType, nullable = false)
val schema = new StructType().add("struct", innerFields)
Expand Down Expand Up @@ -254,6 +265,17 @@ class DataFrameToSchemaSuite extends QueryTest with SharedSparkSession {
checkAnswer(df, Row(Seq(Row(1L))))
}

test("array element: compatible field nullability") {
val innerFields = ArrayType(LongType, containsNull = false)
val schema = new StructType().add("a", LongType).add("b", innerFields)
val data = sql("VALUES (1, ARRAY(1, 2)), (NULL, NULL) as t(a, b)")
assert(data.schema.fields(1).nullable)
assert(!data.schema.fields(1).dataType.asInstanceOf[ArrayType].containsNull)
val df = data.to(schema)
assert(df.schema == schema)
checkAnswer(df, Seq(Row(1, Seq(1, 2)), Row(null, null)))
}

test("array element: incompatible array nullability") {
val arr = ArrayType(IntegerType, containsNull = false)
val schema = new StructType().add("arr", arr)
Expand Down Expand Up @@ -321,6 +343,17 @@ class DataFrameToSchemaSuite extends QueryTest with SharedSparkSession {
checkAnswer(df, Row(Map("a" -> Row("b", "a"))))
}

test("map value: compatible field nullability") {
val innerFields = MapType(StringType, LongType, valueContainsNull = false)
val schema = new StructType().add("a", LongType).add("b", innerFields)
val data = sql("VALUES (1, MAP('a', 1, 'b', 2)), (NULL, NULL) as t(a, b)")
assert(data.schema.fields(1).nullable)
assert(!data.schema.fields(1).dataType.asInstanceOf[MapType].valueContainsNull)
val df = data.to(schema)
assert(df.schema == schema)
checkAnswer(df, Seq(Row(1, Map("a" -> 1, "b" -> 2)), Row(null, null)))
}

test("map value: incompatible map nullability") {
val m = MapType(StringType, StringType, valueContainsNull = false)
val schema = new StructType().add("map", m, nullable = false)
Expand Down

0 comments on commit a1b853c

Please sign in to comment.