GH-48455: [Python] Handle nested field names when sanitizing table at ParquetWriter (flavor='spark')

python/pyarrow/parquet/core.py

@@ -715,32 +715,49 @@ def _sanitized_spark_field_name(name):
     return _SPARK_DISALLOWED_CHARS.sub('_', name)
 
 
-def _sanitize_schema(schema, flavor):
-    if 'spark' in flavor:
-        sanitized_fields = []
+def _sanitize_field_recursive(field):
+    """
+    Recursively sanitize field names in struct types for Spark compatibility.
 
-        schema_changed = False
+    Returns
+    -------
+    tuple
+        (sanitized_field, changed) where changed is True if any sanitization occurred
+    """
+    sanitized_name = _sanitized_spark_field_name(field.name)
+    sanitized_type = field.type
+    type_changed = False
 
-        for field in schema:
-            name = field.name
-            sanitized_name = _sanitized_spark_field_name(name)
+    if pa.types.is_struct(field.type):
+        sanitized_fields = [_sanitize_field_recursive(f) for f in field.type]
+        if any(changed for _, changed in sanitized_fields):
+            sanitized_type = pa.struct([f for f, _ in sanitized_fields])
+            type_changed = True
 
-            if sanitized_name != name:
-                schema_changed = True
-                sanitized_field = pa.field(sanitized_name, field.type,
-                                           field.nullable, field.metadata)
-                sanitized_fields.append(sanitized_field)
-            else:
-                sanitized_fields.append(field)
+    name_changed = sanitized_name != field.name
+    if name_changed or type_changed:
+        return pa.field(sanitized_name, sanitized_type, field.nullable,
+                        field.metadata), True
+    return field, False
 
-        new_schema = pa.schema(sanitized_fields, metadata=schema.metadata)
-        return new_schema, schema_changed
-    else:
+
+def _sanitize_schema(schema, flavor):
+    if 'spark' not in flavor:
         return schema, False
 
+    sanitized_fields = []
+    schema_changed = False
+
+    for field in schema:
+        sanitized_field, changed = _sanitize_field_recursive(field)
+        sanitized_fields.append(sanitized_field)
+        schema_changed = schema_changed or changed
+
+    new_schema = pa.schema(sanitized_fields, metadata=schema.metadata)
+    return new_schema, schema_changed
+
 
 def _sanitize_table(table, new_schema, flavor):
-    # TODO: This will not handle prohibited characters in nested field names
     if 'spark' in flavor:
         column_data = [table[i] for i in range(table.num_columns)]
         return pa.Table.from_arrays(column_data, schema=new_schema)

python/pyarrow/tests/parquet/test_basic.py

@@ -613,14 +613,49 @@ def test_compression_level():
 
 
 def test_sanitized_spark_field_names():
-    a0 = pa.array([0, 1, 2, 3, 4])
-    name = 'prohib; ,\t{}'
-    table = pa.Table.from_arrays([a0], [name])
-
+    field_metadata = {b'key': b'value'}
+    schema_metadata = {b'schema_key': b'schema_value'}
+
+    schema = pa.schema([
+        pa.field('prohib; ,\t{}', pa.int32()),
+        pa.field('field=with\nspecial', pa.string(), metadata=field_metadata),
+        pa.field('nested_struct', pa.struct([
+            pa.field('field,comma', pa.int32()),
+            pa.field('deeply{nested}', pa.struct([
+                pa.field('field(parens)', pa.float64()),
+                pa.field('normal_field', pa.bool_())
+            ]))
+        ]))
+    ], metadata=schema_metadata)
+
+    data = [
+        pa.array([1, 2]),
+        pa.array(['a', 'b']),
+        pa.array([
+            {'field,comma': 10, 'deeply{nested}': {
+                'field(parens)': 1.5, 'normal_field': True}},
+            {'field,comma': 20, 'deeply{nested}': {
+                'field(parens)': 2.5, 'normal_field': False}}
+        ], type=schema[2].type)
+    ]
+
+    table = pa.Table.from_arrays(data, schema=schema)
     result = _roundtrip_table(table, write_table_kwargs={'flavor': 'spark'})
 
-    expected_name = 'prohib______'
-    assert result.schema[0].name == expected_name
+    assert result.schema[0].name == 'prohib______'
+    assert result.schema[1].name == 'field_with_special'
+
+    nested_type = result.schema[2].type
+    assert nested_type[0].name == 'field_comma'
+    assert nested_type[1].name == 'deeply_nested_'
+
+    deep_type = nested_type[1].type
+    assert deep_type[0].name == 'field_parens_'
+    assert deep_type[1].name == 'normal_field'
+
+    assert result.schema[1].metadata == field_metadata
+    assert result.schema.metadata == schema_metadata
+    assert len(result) == 2
 
 
 @pytest.mark.pandas