code: def write_to_dataset(
      df,
      root_path,
      compression="snappy",
      filename=None,
      partition_cols=None,
      fs=None,
      preserve_index=False,
      return_metadata=False,
      statistics="ROWGROUP",
      int96_timestamps=False,
      row_group_size_bytes=ioutils._ROW_GROUP_SIZE_BYTES_DEFAULT,
      row_group_size_rows=None,
      max_page_size_bytes=None,
      max_page_size_rows=None,
      storage_options=None,
      force_nullable_schema=False,
  ):
      """Wraps `to_parquet` to write partitioned Parquet datasets.
      For each combination of partition group and value,
      subdirectories are created as follows:
  
      .. code-block:: bash
  
          root_dir/
              group=value1
                  <filename>.parquet
              ...
              group=valueN
                  <filename>.parquet
  
      Parameters
      ----------
      df : cudf.DataFrame
      root_path : string,
          The root directory of the dataset
      compression : {'snappy', 'ZSTD', None}, default 'snappy'
          Name of the compression to use. Use ``None`` for no compression.
      filename : string, default None
          The file name to use (within each partition directory). If None,
          a random uuid4 hex string will be used for each file name.
      partition_cols : list,
          Column names by which to partition the dataset.
          Columns are partitioned in the order they are given.
      fs : FileSystem, default None
          If nothing passed, paths assumed to be found in the local on-disk
          filesystem
      preserve_index : bool, default False
          Preserve index values in each parquet file.
      return_metadata : bool, default False
          Return parquet metadata for written data. Returned metadata will
          include the file-path metadata (relative to `root_path`).
      int96_timestamps : bool, default False
          If ``True``, write timestamps in int96 format. This will convert
          timestamps from timestamp[ns], timestamp[ms], timestamp[s], and
          timestamp[us] to the int96 format, which is the number of Julian
          days and the number of nanoseconds since midnight of 1970-01-01.
          If ``False``, timestamps will not be altered.
      row_group_size_bytes: integer or None, default None
          Maximum size of each stripe of the output.
          If None, 134217728 (128MB) will be used.
      row_group_size_rows: integer or None, default None
          Maximum number of rows of each stripe of the output.
          If None, 1000000 will be used.
      max_page_size_bytes: integer or None, default None
          Maximum uncompressed size of each page of the output.
          If None, 524288 (512KB) will be used.
      max_page_size_rows: integer or None, default None
          Maximum number of rows of each page of the output.
          If None, 20000 will be used.
      storage_options : dict, optional, default None
          Extra options that make sense for a particular storage connection,
          e.g. host, port, username, password, etc. For HTTP(S) URLs the
          key-value pairs are forwarded to ``urllib.request.Request`` as
          header options. For other URLs (e.g. starting with "s3://", and
          "gcs://") the key-value pairs are forwarded to ``fsspec.open``.
          Please see ``fsspec`` and ``urllib`` for more details.
      force_nullable_schema : bool, default False.
          If True, writes all columns as `null` in schema.
          If False, columns are written as `null` if they contain null values,
          otherwise as `not null`.
      """
  
      fs = ioutils._ensure_filesystem(fs, root_path, storage_options)
      fs.mkdirs(root_path, exist_ok=True)
  
      if partition_cols is not None and len(partition_cols) > 0:
          (
              full_paths,
              metadata_file_paths,
              grouped_df,
              part_offsets,
              _,
          ) = _get_partitioned(
              df=df,
              root_path=root_path,
              partition_cols=partition_cols,
              filename=filename,
              fs=fs,
              preserve_index=preserve_index,
              storage_options=storage_options,
          )
          metadata_file_path = (
              metadata_file_paths if return_metadata else None
          )
          metadata = to_parquet(
              df=grouped_df,
              path=full_paths,
              compression=compression,
              index=preserve_index,
              partition_offsets=part_offsets,
              storage_options=storage_options,
              metadata_file_path=metadata_file_path,
              statistics=statistics,
              int96_timestamps=int96_timestamps,
              row_group_size_bytes=row_group_size_bytes,
              row_group_size_rows=row_group_size_rows,
              max_page_size_bytes=max_page_size_bytes,
              max_page_size_rows=max_page_size_rows,
              force_nullable_schema=force_nullable_schema,
          )
  
      else:
          filename = filename or _generate_filename()
          full_path = fs.sep.join([root_path, filename])
  
          metadata_file_path = filename if return_metadata else None
  
          metadata = df.to_parquet(
              path=full_path,
              compression=compression,
              index=preserve_index,
              storage_options=storage_options,
              metadata_file_path=metadata_file_path,
              statistics=statistics,
              int96_timestamps=int96_timestamps,
              row_group_size_bytes=row_group_size_bytes,
              row_group_size_rows=row_group_size_rows,
              max_page_size_bytes=max_page_size_bytes,
              max_page_size_rows=max_page_size_rows,
              force_nullable_schema=force_nullable_schema,
          )
  
      return metadata
  
imports: from __future__ import annotations
  import itertools
  import math
  import operator
  import shutil
  import tempfile
  import warnings
  from collections import defaultdict
  from contextlib import ExitStack
  from functools import partial, reduce
  from typing import Callable, Dict, List, Optional, Tuple
  from uuid import uuid4
  import numpy
  import pandas
  from pyarrow import dataset
  import cudf
  from cudf._lib import parquet
  from cudf.api.types import is_list_like
  from cudf.core.column import build_categorical_column, column_empty, full
  from cudf.utils import ioutils
  from cudf.utils.nvtx_annotation import _cudf_nvtx_annotate
label_desc: Use image node_id 2 for calling a global function with example usage: write_to_dataset(df, root_path, compression, filename, partition_cols, fs, preserve_index, return_metadata, statistics, int96_timestamps, row_group_size_bytes, row_group_size_rows, max_page_size_bytes, max_page_size_rows, storage_options, force_nullable_schema) and returns: metadata