write with sanitized column names

Author: kevinjqliu

pyiceberg/io/pyarrow.py

@@ -1761,7 +1761,7 @@ def data_file_statistics_from_parquet_metadata(
 
 
 def write_file(io: FileIO, table_metadata: TableMetadata, tasks: Iterator[WriteTask]) -> Iterator[DataFile]:
-    schema = table_metadata.schema()
+    schema = sanitize_column_names(table_metadata.schema())
     arrow_file_schema = schema.as_arrow()
     parquet_writer_kwargs = _get_parquet_writer_kwargs(table_metadata.properties)
 

pyiceberg/table/__init__.py

@@ -92,6 +92,7 @@
     SchemaWithPartnerVisitor,
     assign_fresh_schema_ids,
     promote,
+    sanitize_column_names,
     visit,
     visit_with_partner,
 )
@@ -2702,6 +2703,8 @@ def _dataframe_to_data_files(
         property_name=TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
         default=TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT,
     )
+    sanitized_arrow_schema = sanitize_column_names(table_metadata.schema()).as_arrow()
+    df = df.rename_columns(sanitized_arrow_schema.names)
 
     if len(table_metadata.spec().fields) > 0:
         partitions = _determine_partitions(spec=table_metadata.spec(), schema=table_metadata.schema(), arrow_table=df)