ENH: pass through schema keyword in to_parquet for pyarrow

doc/source/whatsnew/v1.0.0.rst

@@ -798,7 +798,6 @@ I/O
 - Bug in :func:`read_json` where default encoding was not set to ``utf-8`` (:issue:`29565`)
 - Bug in :class:`PythonParser` where str and bytes were being mixed when dealing with the decimal field (:issue:`29650`)
 - :meth:`read_gbq` now accepts ``progress_bar_type`` to display progress bar while the data downloads. (:issue:`29857`)
--
 
 Plotting
 ^^^^^^^^
@@ -880,7 +879,7 @@ Other
 - Fixed :class:`IntegerArray` returning ``inf`` rather than ``NaN`` for operations dividing by 0 (:issue:`27398`)
 - Fixed ``pow`` operations for :class:`IntegerArray` when the other value is ``0`` or ``1`` (:issue:`29997`)
 - Bug in :meth:`Series.count` raises if use_inf_as_na is enabled (:issue:`29478`)
-
+- :meth: `to_parquet` now appropriately handles the ``schema`` argument for user defined schemas in the pyarrow engine. (:issue: `30270`)
 
 .. _whatsnew_1000.contributors:
 

pandas/io/parquet.py

@@ -91,11 +91,10 @@ def write(
         self.validate_dataframe(df)
         path, _, _, _ = get_filepath_or_buffer(path, mode="wb")
 
-        from_pandas_kwargs: Dict[str, Any]
-        if index is None:
-            from_pandas_kwargs = {}
-        else:
-            from_pandas_kwargs = {"preserve_index": index}
+        from_pandas_kwargs: Dict[str, Any] = {"schema": kwargs.pop("schema", None)}
+        if index is not None:
+            from_pandas_kwargs["preserve_index"] = index
+
         table = self.api.Table.from_pandas(df, **from_pandas_kwargs)
         if partition_cols is not None:
             self.api.parquet.write_to_dataset(

pandas/tests/io/test_parquet.py

@@ -517,6 +517,14 @@ def test_empty_dataframe(self, pa):
         df = pd.DataFrame()
         check_round_trip(df, pa)
 
+    def test_write_with_schema(self, pa):
+        import pyarrow
+
+        df = pd.DataFrame({"x": [0, 1]})
+        schema = pyarrow.schema([pyarrow.field("x", type=pyarrow.bool_())])
+        out_df = df.astype(bool)
+        check_round_trip(df, pa, write_kwargs={"schema": schema}, expected=out_df)
+
     @pytest.mark.skip(reason="broken test")
     @td.skip_if_no("pyarrow", min_version="0.15.0")
     def test_additional_extension_arrays(self, pa):