[SPARK-61146][CONNECT][PYTHON] Create DataFrames of pandas frames whose struct fields repeat names in the Python client, as PySpark does

crates/pyspark-rs/src/session.rs

@@ -776,14 +776,21 @@ impl PySparkSession {
             safecheck,
             large,
         ))?;
-        let frame = crate::pyarrow_import::create_dataframe(
-            py,
-            &self.session,
-            &table,
-            Some(schema_py.bind(py)),
-        )?;
+        // As PySpark Connect, the table, cast to the Arrow types of the schema of deduplicated
+        // nested field names, is sent with the schema as given, which the server applies. It is
+        // not normalized again as an Arrow table of the caller, which may not repeat the names
+        // of the fields of a struct (DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT).
+        let spark_schema = crate::types::py_to_data_type(schema_py.bind(py))?;
+        let batches = crate::pyarrow_import::table_batches(&table)?;
+        let df = py
+            .detach(|| {
+                self.session
+                    .create_dataframe_from_arrow_with_options(batches, Some(spark_schema), &options)
+                    .map_err(Box::new)
+            })
+            .to_pyerr()?;
         // As the other pandas paths, the explicit names are applied with toDF.
-        Ok(Some(PyDataFrame::new(frame.dataframe.to_df(
+        Ok(Some(PyDataFrame::new(df.to_df(
             fields.iter().map(|field| field.name.as_str()).collect(),
         ))))
     }

python/tests/integer_tuple_cases.py

@@ -145,8 +145,9 @@ def capture(self, data, schema):
         frame = self.spark.createDataFrame(data, schema)
         self.assertEqual(len(self.captured), before)  # construction remains offline
         # Explicit-schema pandas input reads the Arrow conversion settings during
-        # construction, as PySpark does: the normalization and safe-cast configuration.
-        expected = 3 if isinstance(data, pd.DataFrame) else 0
+        # construction, as PySpark does: the Arrow input settings and the safe-cast
+        # configuration, once each.
+        expected = 2 if isinstance(data, pd.DataFrame) else 0
         self.assertEqual(self.config_calls - config_before, expected)
         config_before = self.config_calls
         frame.toArrow()

python/tests/test_dropin_e2e.py

@@ -2470,6 +2470,33 @@ def test_struct_types_convert_to_the_ddl_of_spark_as_pyspark(session):
 
 
 
+def test_pandas_frames_of_duplicate_nested_field_names_are_created_as_pyspark(session):
+    # As PySpark's test_createDataFrame_pandas_duplicate_field_names: the structs of the schema
+    # repeat field names, with Arrow and without.
+    import pandas as pd
+    from pyspark.sql.types import StructType
+
+    schema = (
+        StructType()
+        .add("struct", StructType().add("x", StringType()).add("x", IntegerType()))
+        .add(
+            "struct",
+            StructType().add("a", IntegerType()).add("x", IntegerType()).add("x", StringType()),
+        )
+    )
+    data = [Row(Row("a", 1), Row(2, 3, "b")), Row(Row("x", None), Row(7, 8, None))]
+    frame = pd.DataFrame.from_records(data, columns=schema.names)
+    try:
+        for arrow in ("true", "false"):
+            session.conf.set("spark.sql.execution.arrow.pyspark.enabled", arrow)
+            df = session.createDataFrame(frame, schema)
+            assert df.schema == schema and df.collect() == data, arrow
+    finally:
+        session.conf.unset("spark.sql.execution.arrow.pyspark.enabled")
+
+
+
+
 def test_maps_keep_their_entries_in_order_with_keys_of_their_type(session):
     row = session.range(1).selectExpr(
         "map(2, 'b', 1, 'a') AS ints",

python/tests/test_dropin_offline.py

@@ -6943,6 +6943,30 @@ def test_local_input_native_number_values_and_types(mock_connect, value):
         assert actual == value
 
 
+def test_pandas_frames_of_duplicate_nested_field_names_are_sent_as_pyspark(mock_connect):
+    # As PySpark Connect: a pandas frame under a schema whose structs repeat field names is sent
+    # as the Arrow table of the deduplicated names (x_0, x_1), with the schema as given, which
+    # the server applies. Before, the client raised DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT.
+    pd = pytest.importorskip("pandas")
+    pa = pytest.importorskip("pyarrow")
+
+    spark, service = mock_connect
+    schema = T.StructType().add(
+        "s", T.StructType().add("x", T.StringType()).add("x", T.IntegerType())
+    )
+    frame = pd.DataFrame.from_records([(Row("a", 1),), (Row("b", None),)], columns=schema.names)
+    spark.createDataFrame(frame, schema).toArrow()
+    relation = parse_message(parse_message(parse_message(service.execute_requests[-1])[3][0])[1][0])
+    while 18 in relation:  # Relation.to_df -> ToDF.input.
+        relation = parse_message(parse_message(relation[18][0])[1][0])
+    # Relation.local_relation=11 -> LocalRelation.data=1, schema=2.
+    local = parse_message(relation[11][0])
+    table = pa.ipc.open_stream(local[1][0]).read_all()
+    assert table.schema.field("s").type == pa.struct([("x_0", pa.string()), ("x_1", pa.int32())])
+    assert table.to_pylist() == [{"s": {"x_0": "a", "x_1": 1}}, {"s": {"x_0": "b", "x_1": None}}]
+    assert T.StructType.fromJson(json.loads(local[2][0])) == schema
+
+
 def test_pandas_input_overflow_raises_as_pyspark(mock_connect):
     # Without a schema, PySpark infers the Arrow type of an object column, so an integer
     # outside the int64 range raises OverflowError rather than becoming a double.