diff --git a/crates/pyspark-rs/src/session.rs b/crates/pyspark-rs/src/session.rs index 1d822455..ecdf526a 100644 --- a/crates/pyspark-rs/src/session.rs +++ b/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(), )))) } diff --git a/python/tests/integer_tuple_cases.py b/python/tests/integer_tuple_cases.py index ed49ae29..aa4fce94 100644 --- a/python/tests/integer_tuple_cases.py +++ b/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() diff --git a/python/tests/test_dropin_e2e.py b/python/tests/test_dropin_e2e.py index e70e9fbe..3ed9f766 100644 --- a/python/tests/test_dropin_e2e.py +++ b/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", diff --git a/python/tests/test_dropin_offline.py b/python/tests/test_dropin_offline.py index a8d89486..31920a94 100644 --- a/python/tests/test_dropin_offline.py +++ b/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.