[SPARK-35929][PYTHON] Support to infer nested dict as a struct when creating a DataFrame

python/pyspark/sql/tests/test_types.py

@@ -196,6 +196,12 @@ def test_infer_nested_schema(self):
         df = self.spark.createDataFrame(nestedRdd2)
         self.assertEqual(Row(f1=[[1, 2], [2, 3]], f2=[1, 2]), df.collect()[0])
 
+        with self.sql_conf({"spark.sql.pyspark.inferNestedStructByMap": False}):
+            nestedRdd3 = self.sc.parallelize([NestedRow([{"payment": 200.5, "name": "A"}], [1, 2]),
+                                              NestedRow([{"payment": 100.5, "name": "B"}], [2, 3])])
+            df = self.spark.createDataFrame(nestedRdd3)
+            self.assertEqual(Row(f1=[Row(payment=200.5, name='A')], f2=[1, 2]), df.collect()[0])
+
         from collections import namedtuple
         CustomRow = namedtuple('CustomRow', 'field1 field2')
         rdd = self.sc.parallelize([CustomRow(field1=1, field2="row1"),

python/pyspark/sql/types.py

@@ -1020,10 +1020,19 @@ def _infer_type(obj):
         return dataType()
 
     if isinstance(obj, dict):
-        for key, value in obj.items():
-            if key is not None and value is not None:
-                return MapType(_infer_type(key), _infer_type(value), True)
-        return MapType(NullType(), NullType(), True)
+        from pyspark.sql.session import SparkSession
+        if (SparkSession._activeSession.conf.get(
+                "spark.sql.pyspark.inferNestedStructByMap").lower() == "true"):
+            for key, value in obj.items():
+                if key is not None and value is not None:
+                    return MapType(_infer_type(key), _infer_type(value), True)
+            return MapType(NullType(), NullType(), True)
+        else:
+            struct = StructType()
+            for key, value in obj.items():
+                if key is not None and value is not None:
+                    struct.add(key, _infer_type(value), True)
+            return struct
     elif isinstance(obj, list):
         for v in obj:
             if v is not None:

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3327,6 +3327,13 @@ object SQLConf {
     .intConf
     .createWithDefault(0)
 
+  val INFER_NESTED_STRUCT_BY_MAP = buildConf("spark.sql.pyspark.inferNestedStructByMap")
+    .internal()
+    .doc("When set to false, inferring the nested struct by StructType. MapType is default.")
+    .version("3.2.0")
+    .booleanConf
+    .createWithDefault(true)
+
   /**
    * Holds information about keys that have been deprecated.
    *
@@ -4040,6 +4047,8 @@ class SQLConf extends Serializable with Logging {
 
   def maxConcurrentOutputFileWriters: Int = getConf(SQLConf.MAX_CONCURRENT_OUTPUT_FILE_WRITERS)
 
+  def inferNestedStructByMap: Boolean = getConf(SQLConf.INFER_NESTED_STRUCT_BY_MAP)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */