Checks row length when converting Java arrays to Python rows

Author: liancheng

python/pyspark/sql/tests.py

@@ -364,6 +364,15 @@ def test_infer_schema_to_local(self):
         df3 = self.sqlCtx.createDataFrame(rdd, df.schema)
         self.assertEqual(10, df3.count())
 
+    def test_create_dataframe_schema_mismatch(self):
+        input = [Row(a=1)]
+        rdd = self.sc.parallelize(range(3)).map(lambda i: Row(a=i))
+        schema = StructType([StructField("a", IntegerType()), StructField("b", StringType())])
+        df = self.sqlCtx.createDataFrame(rdd, schema)
+        message = ".*assertion failed: Row length 1 and schema length 2 don't match.*"
+        with self.assertRaisesRegexp(Exception, message):
+            df.show()
+
     def test_serialize_nested_array_and_map(self):
         d = [Row(l=[Row(a=1, b='s')], d={"key": Row(c=1.0, d="2")})]
         rdd = self.sc.parallelize(d)

sql/core/src/main/scala/org/apache/spark/sql/execution/python.scala

@@ -220,7 +220,12 @@ object EvaluatePython {
       ArrayBasedMapData(keys, values)
 
     case (c, StructType(fields)) if c.getClass.isArray =>
-      new GenericInternalRow(c.asInstanceOf[Array[_]].zip(fields).map {
+      val array = c.asInstanceOf[Array[_]]
+      assert(
+        array.length == fields.length,
+        s"Row length ${array.length} and schema length ${fields.length} don't match"
+      )
+      new GenericInternalRow(array.zip(fields).map {
         case (e, f) => fromJava(e, f.dataType)
       })