Add utc_timestamps parameter to cudf.io.parquet.ParquetWriter

Behavioral change: allow chunked writes to emit timestamp columns with
isAdjustedToUTC=false, as DataFrame.to_parquet does. The default (True)
keeps the current output unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: a-hirota

python/cudf/cudf/io/parquet.py

@@ -1858,6 +1858,10 @@ class ParquetWriter:
     store_schema : bool, default False
         If ``True``, enable computing and writing arrow schema to Parquet
         file footer's key-value metadata section for faithful round-tripping.
+    utc_timestamps : bool, default True
+        If ``True``, timestamp columns are written with
+        ``isAdjustedToUTC=true``. If ``False``, they are written with
+        ``isAdjustedToUTC=false``, matching ``DataFrame.to_parquet``.
 
     See Also
     --------
@@ -1877,6 +1881,7 @@ def __init__(
         max_dictionary_size: int = 1048576,
         use_dictionary: bool = True,
         store_schema: bool = False,
+        utc_timestamps: bool = True,
     ):
         filepaths_or_buffers = (
             list(filepath_or_buffer)
@@ -1895,6 +1900,7 @@ def __init__(
         self.max_dictionary_size = max_dictionary_size
         self.use_dictionary = use_dictionary
         self.write_arrow_schema = store_schema
+        self.utc_timestamps = utc_timestamps
 
     def write_table(self, table, partitions_info=None) -> None:
         """Writes a single table to the file"""
@@ -2001,6 +2007,7 @@ def _initialize_chunked_state(
             .max_page_size_rows(self.max_page_size_rows)
             .max_dictionary_size(self.max_dictionary_size)
             .write_arrow_schema(self.write_arrow_schema)
+            .utc_timestamps(self.utc_timestamps)
             .build()
         )
         options.set_dictionary_policy(dict_policy)

python/cudf/cudf/tests/input_output/test_parquet.py

@@ -5,6 +5,7 @@
 import decimal
 import glob
 import hashlib
+import json
 import math
 import os
 import pathlib
@@ -1706,6 +1707,28 @@ def test_parquet_writer_int96_timestamps(tmp_path, pdf, gdf):
     assert_eq(expect, got, check_categorical=False, check_dtype=False)
 
 
+@pytest.mark.parametrize(
+    "kwargs, expected", [({}, True), ({"utc_timestamps": False}, False)]
+)
+def test_parquet_writer_chunked_utc_timestamps(tmp_path, kwargs, expected):
+    gdf = cudf.DataFrame(
+        {"ts": cudf.Series([0, 1_000_000, None], dtype="datetime64[us]")}
+    )
+    fname = tmp_path / "gdf.parquet"
+
+    with ParquetWriter(fname, **kwargs) as writer:
+        writer.write_table(gdf)
+        writer.write_table(gdf)
+
+    logical_type = json.loads(
+        pq.ParquetFile(fname).schema.column(0).logical_type.to_json()
+    )
+    assert logical_type["isAdjustedToUTC"] is expected
+    assert_eq(
+        cudf.read_parquet(fname), cudf.concat([gdf, gdf], ignore_index=True)
+    )
+
+
 def test_multifile_parquet_folder(tmp_path):
     test_pdf1 = pd.DataFrame(
         {