fix: forward write_options when write_parquet receives ParquetWriterOptions (#1761)

`write_parquet` takes `write_options`, documents it and declares it in
the `@overload` for the `ParquetWriterOptions` form, but that branch
calls `write_parquet_with_options(path, compression)` and drops it. The
destination accepts the parameter, so everything in
`DataFrameWriteOptions` (`partition_by`, `single_file_output`,
`insert_operation`, `sort_by`) was silently ignored for that one
spelling.

Measured with the same `DataFrameWriteOptions(partition_by="part")`:

    write_parquet(path, ParquetWriterOptions(), write_options=wo)
        -> ['IDuOjvMa3pdEDotb_0.parquet']     not partitioned
    write_parquet_with_options(path, ParquetWriterOptions(), write_options=wo)
        -> ['part=a', 'part=b']
    write_parquet(path, "zstd", write_options=wo)
        -> ['part=a', 'part=b']

No error and no warning: the files just land in the wrong layout.

The branch arrived in ef62fa8 (#1169) while `write_options` came
earlier in #857, so the new delegation path was written without carrying
the existing parameter over. The same `if` refuses `compression_level`
with an explicit `ValueError`, which shows that arguments incompatible
with this branch get rejected on purpose; `write_options` was not
rejected, only forgotten.

Author: Rodrigo-Palma

python/datafusion/dataframe.py

@@ -1621,7 +1621,7 @@ def write_parquet(
             if compression_level is not None:
                 msg = "compression_level should be None when using ParquetWriterOptions"
                 raise ValueError(msg)
-            self.write_parquet_with_options(path, compression)
+            self.write_parquet_with_options(path, compression, write_options)
             return
 
         if isinstance(compression, str):

python/tests/test_dataframe.py

@@ -2538,6 +2538,25 @@ def test_write_parquet(df, tmp_path, path_to_str):
     assert result == expected
 
 
+def test_write_parquet_writer_options_keeps_write_options(ctx, tmp_path):
+    """``write_parquet`` honours ``write_options`` alongside ``ParquetWriterOptions``.
+
+    The ``ParquetWriterOptions`` branch delegates to
+    :py:meth:`DataFrame.write_parquet_with_options`, which takes ``write_options``
+    too, so ``partition_by`` must still reach the writer.
+    """
+    df = ctx.from_pydict({"part": ["a", "a", "b"], "v": [1, 2, 3]})
+    path = tmp_path / "partitioned"
+
+    df.write_parquet(
+        path,
+        ParquetWriterOptions(),
+        write_options=DataFrameWriteOptions(partition_by="part"),
+    )
+
+    assert sorted(p.name for p in path.iterdir()) == ["part=a", "part=b"]
+
+
 @pytest.mark.parametrize(
     ("compression", "compression_level"),
     [("gzip", 6), ("brotli", 7), ("zstd", 15)],