Validated data frames with not null columns against null schema columns.

Previously, a dataframe would be considered invalid if it had not-null data, but the schema allowed the data to be nullable
mrpowers-io · Jun 10, 2021 · a90b4bb · a90b4bb
1 parent 791606f
commit a90b4bb
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 3 deletions.
diff --git a/src/main/scala/com/github/mrpowers/spark/daria/sql/DataFrameSchemaChecker.scala b/src/main/scala/com/github/mrpowers/spark/daria/sql/DataFrameSchemaChecker.scala
@@ -18,17 +18,17 @@ private[sql] class DataFrameSchemaChecker(df: DataFrame, requiredSchema: StructT
       case Success(namedField) =>
         val basicMatch =
           namedField.name == reqField.name &&
-            namedField.nullable == reqField.nullable &&
+            (!namedField.nullable || reqField.nullable) &&
             namedField.metadata == reqField.metadata
-
         val contentMatch = reqField.dataType match {
           case reqSchema: StructType =>
             namedField.dataType match {
               case fieldSchema: StructType =>
                 diff(reqSchema, fieldSchema).isEmpty
               case _ => false
             }
-          case _ => reqField == namedField
+          case namedField.dataType => true
+          case _                   => false
         }
 
         basicMatch && contentMatch

diff --git a/src/test/scala/com/github/mrpowers/spark/daria/sql/DataFrameSchemaCheckerTest.scala b/src/test/scala/com/github/mrpowers/spark/daria/sql/DataFrameSchemaCheckerTest.scala
@@ -363,6 +363,40 @@ object DataFrameSchemaCheckerTest extends TestSuite with SparkSessionTestWrapper
 
       }
 
+      "validates non-null column against a null schema" - {
+        val sourceSchema = List(
+          StructField(
+            "num",
+            IntegerType,
+            false
+          )
+        )
+
+        val sourceDF =
+          spark.createDataFrame(
+            spark.sparkContext.parallelize(Seq[Row]()),
+            StructType(sourceSchema)
+          )
+
+        val requiredSchema =
+          StructType(
+            List(
+              StructField(
+                "num",
+                IntegerType,
+                true
+              )
+            )
+          )
+
+        val c = new DataFrameSchemaChecker(
+          sourceDF,
+          requiredSchema
+        )
+
+        c.validateSchema()
+      }
+
     }
 
   }