[SPARK-42073][CONNECT][PYTHON][TESTS] Enable tests in common/test_parity_serde, common/test_parity_types

### What changes were proposed in this pull request?
Enable tests in common/test_parity_serde, common/test_parity_types

### Why are the changes needed?
for test coverage

### Does this PR introduce _any_ user-facing change?
no, test-only

### How was this patch tested?
enabled UT

Closes #39583 from techaddict/SPARK-42073.

Authored-by: Sandeep Singh <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: techaddict

python/pyspark/sql/tests/connect/test_parity_serde.py

@@ -30,10 +30,6 @@ def test_int_array_serialization(self):
     def test_serialize_nested_array_and_map(self):
         super().test_serialize_nested_array_and_map()
 
-    @unittest.skip("Spark Connect does not support RDD but the tests depend on them.")
-    def test_struct_in_map(self):
-        super().test_struct_in_map()
-
 
 if __name__ == "__main__":
     import unittest

python/pyspark/sql/tests/connect/test_parity_types.py

@@ -59,10 +59,6 @@ def test_cast_to_udt_with_udt(self):
     def test_complex_nested_udt_in_df(self):
         super().test_complex_nested_udt_in_df()
 
-    @unittest.skip("Spark Connect does not support RDD but the tests depend on them.")
-    def test_convert_row_to_dict(self):
-        super().test_convert_row_to_dict()
-
     # TODO(SPARK-42020): createDataFrame with UDT
     @unittest.skip("Fails in Spark Connect, should enable.")
     def test_create_dataframe_from_objects(self):
@@ -137,10 +133,6 @@ def test_infer_schema_upcast_int_to_string(self):
     def test_infer_schema_with_udt(self):
         super().test_infer_schema_with_udt()
 
-    @unittest.skip("Spark Connect does not support RDD but the tests depend on them.")
-    def test_metadata_null(self):
-        super().test_metadata_null()
-
     # TODO(SPARK-41834): Implement SparkSession.conf
     @unittest.skip("Fails in Spark Connect, should enable.")
     def test_negative_decimal(self):

python/pyspark/sql/tests/test_serde.py

@@ -54,7 +54,7 @@ def test_select_null_literal(self):
 
     def test_struct_in_map(self):
         d = [Row(m={Row(i=1): Row(s="")})]
-        df = self.sc.parallelize(d).toDF()
+        df = self.spark.createDataFrame(d)
         k, v = list(df.head().m.items())[0]
         self.assertEqual(1, k.i)
         self.assertEqual("", v.s)

python/pyspark/sql/tests/test_types.py

@@ -480,7 +480,7 @@ def test_apply_schema(self):
     def test_convert_row_to_dict(self):
         row = Row(l=[Row(a=1, b="s")], d={"key": Row(c=1.0, d="2")})
         self.assertEqual(1, row.asDict()["l"][0].a)
-        df = self.sc.parallelize([row]).toDF()
+        df = self.spark.createDataFrame([row])
 
         with self.tempView("test"):
             df.createOrReplaceTempView("test")
@@ -788,8 +788,7 @@ def test_metadata_null(self):
                 StructField("f2", StringType(), True, {"a": None}),
             ]
         )
-        rdd = self.sc.parallelize([["a", "b"], ["c", "d"]])
-        self.spark.createDataFrame(rdd, schema)
+        self.spark.createDataFrame([["a", "b"], ["c", "d"]], schema)
 
     def test_access_nested_types(self):
         df = self.spark.createDataFrame([Row(l=[1], r=Row(a=1, b="b"), d={"k": "v"})])