fix: preserve column names with spaces in wr.redshift.copy() (#3298)

* fix: preserve column names with spaces in wr.redshift.copy()

Passes flavor=None to internal s3.to_parquet call to prevent pyarrow spark flavor from sanitizing column names (spaces → underscores). Fixes #3293

* feat: add sanitize_column_names parameter to wr.redshift.copy()

Problem:
wr.redshift.copy() internally calls s3.to_parquet() which defaults to
pyarrow flavor='spark'. This causes column names with spaces to be
silently renamed (e.g. "my col" → "my_col"), leading to a mismatch
between the DataFrame schema and the Redshift table schema.

Solution:
Add an optional sanitize_column_names parameter (default=True) to
wr.redshift.copy() that controls whether pyarrow sanitizes column names.

- sanitize_column_names=True (default): preserves existing behavior,
  column names are sanitized for backward compatibility.
- sanitize_column_names=False: passes flavor=None to the internal
  s3.to_parquet() call, preserving original column names including spaces.

This is a non-breaking change — existing users are unaffected since
the default value maintains the current behavior.

Changes:
- Added sanitize_column_names: bool = True parameter to copy()
- Updated pyarrow_additional_kwargs in s3.to_parquet() call accordingly
- Added docstring for the new parameter
- Added test case for sanitize_column_names=False behavior

Fixes #3293

* test: add test for sanitize_column_names=False in wr.redshift.copy()

test: add test for sanitize_column_names=False in wr.redshift.copy()

* style: fix ruff formatting - remove trailing whitespace

style: fix ruff formatting - remove trailing whitespace

* style: fix ruff formatting - remove trailing whitespace in _write.py

style: fix ruff formatting - remove trailing whitespace in _write.py

* style: fix ruff formatting in test_redshift.py

style: fix ruff formatting in test_redshift.py

* fix: add required blank line in docstring for ruff D410/D411

fix: add required blank line in docstring for ruff D410/D411

* fix: remove trailing whitespace in sanitize_column_names docstring

fix: remove trailing whitespace in sanitize_column_names docstring

* fix: replace sanitize_column_names with pyarrow_additional_kwargs

* fix: update test to use pyarrow_additional_kwargs instead of sanitize_column_names

---------

Author: hirenkumar-n-dholariya

awswrangler/redshift/_write.py

@@ -574,6 +574,7 @@ def copy(  # noqa: PLR0913
     precombine_key: str | None = None,
     use_column_names: bool = False,
     add_new_columns: bool = False,
+    pyarrow_additional_kwargs: dict[str, str] | None = None,
 ) -> None:
     """Load Pandas DataFrame as a Table on Amazon Redshift using parquet files on S3 as stage.
 
@@ -687,6 +688,9 @@ def copy(  # noqa: PLR0913
         inserted into the database columns `col1` and `col3`.
     add_new_columns
         If True, it automatically adds the new DataFrame columns into the target table.
+    pyarrow_additional_kwargs
+        Forwarded to pyarrow.
+        e.g. pyarrow_additional_kwargs={'coerce_timestamps': 'us', 'allow_truncated_timestamps': False}
 
     Examples
     --------
@@ -715,6 +719,7 @@ def copy(  # noqa: PLR0913
         s3.to_parquet(
             df=df,
             path=path,
+            pyarrow_additional_kwargs=pyarrow_additional_kwargs or {},
             index=index,
             dataset=True,
             mode="append",

tests/unit/test_redshift.py

@@ -1657,3 +1657,26 @@ def test_copy_serialize_to_json_super(
     df_res["text"] = df_res["text"].apply(json.loads)
 
     assert_pandas_equals(df, df_res)
+
+
+def test_copy_preserve_column_names_with_spaces(
+    path: str,
+    redshift_table: str,
+    redshift_con: "redshift_connector.Connection",
+    databases_parameters: dict[str, Any],
+) -> None:
+    """Test that pyarrow_additional_kwargs={'flavor': None} preserves column names with spaces."""
+    df = pd.DataFrame({"my col": [1, 2, 3], "another col": ["a", "b", "c"]})
+    wr.redshift.copy(
+        df=df,
+        path=path,
+        con=redshift_con,
+        table=redshift_table,
+        schema="public",
+        mode="overwrite",
+        iam_role=databases_parameters["redshift"]["role"],
+        pyarrow_additional_kwargs={"flavor": None},
+    )
+    result = wr.redshift.read_sql_table(table=redshift_table, schema="public", con=redshift_con)
+    assert list(result.columns) == ["my col", "another col"]
+    assert len(result) == 3