Repository: dask/dask
File: dask/dataframe/io/parquet/arrow.py (Python)

Code before:
        if "_metadata" in fns or gather_statistics is not False:
            # Let arrow do its thing (use _metadata or scan files)
            dataset = pq.ParquetDataset(
                paths, filesystem=fs, filters=filters, **dataset_kwargs
            )
        else:
            # Use _common_metadata file if it is available.
            # Otherwise, just use 0th file
            if "_common_metadata" in fns:
                dataset = pq.ParquetDataset(
                    base + fs.sep + "_common_metadata", filesystem=fs, **dataset_kwargs
                )
            else:
                dataset = pq.ParquetDataset(
                    allpaths[0], filesystem=fs, **dataset_kwargs
                )
            parts = [base + fs.sep + fn for fn in fns]
    else:
        # There is only one file to read
        dataset = pq.ParquetDataset(paths, filesystem=fs, **dataset_kwargs)
    return parts, dataset


def _get_filesystem_and_path(passed_filesystem, path):
    if passed_filesystem is None:
        return resolve_filesystem_and_path(path, passed_filesystem)
    return passed_filesystem, path


def _write_partitioned(
    table, root_path, partition_cols, preserve_index=True, filesystem=None, **kwargs
):
    """ Write table to a partitioned dataset with pyarrow.

        Logic copied from pyarrow.parquet.
        (arrow/python/pyarrow/parquet.py::write_to_dataset)
    """
    fs, root_path = _get_filesystem_and_path(filesystem, root_path)
    fs.mkdirs(root_path, exist_ok=True)

    df = table.to_pandas(ignore_metadata=True)
    partition_keys = [df[col] for col in partition_cols]
    data_df = df.drop(partition_cols, axis="columns")
    data_cols = df.columns.drop(partition_cols)
    if len(data_cols) == 0 and not preserve_index:
        raise ValueError("No data left to save outside partition columns")

    subschema = table.schema
    for col in table.schema.names:
        if col in partition_cols:
            subschema = subschema.remove(subschema.get_field_index(col))

Review comment (suggestion):
Again, would be reasonable to explicitly use fsspec here (although in practice, is `passed_filesystem` ever actually `None`?). pyarrow only supports a very limited set of file-system implementations.

Code after:
            # Let arrow do its thing (use _metadata or scan files)
            dataset = pq.ParquetDataset(
                paths, filesystem=fs, filters=filters, **dataset_kwargs
            )
        else:
            # Use _common_metadata file if it is available.
            # Otherwise, just use 0th file
            if "_common_metadata" in fns:
                dataset = pq.ParquetDataset(
                    base + fs.sep + "_common_metadata", filesystem=fs, **dataset_kwargs
                )
            else:
                dataset = pq.ParquetDataset(
                    allpaths[0], filesystem=fs, **dataset_kwargs
                )
            parts = [base + fs.sep + fn for fn in fns]
    else:
        # There is only one file to read
        dataset = pq.ParquetDataset(paths, filesystem=fs, **dataset_kwargs)
    return parts, dataset


def _write_partitioned(
    table, root_path, partition_cols, fs, preserve_index=True, **kwargs
):
    """ Write table to a partitioned dataset with pyarrow.

        Logic copied from pyarrow.parquet.
        (arrow/python/pyarrow/parquet.py::write_to_dataset)

        TODO: Remove this in favor of pyarrow's `write_to_dataset`
              once ARROW-8244 is addressed.
    """
    fs.mkdirs(root_path, exist_ok=True)

    df = table.to_pandas(ignore_metadata=True)
    partition_keys = [df[col] for col in partition_cols]
    data_df = df.drop(partition_cols, axis="columns")
    data_cols = df.columns.drop(partition_cols)
    if len(data_cols) == 0 and not preserve_index:
        raise ValueError("No data left to save outside partition columns")

    subschema = table.schema
    for col in table.schema.names:
        if col in partition_cols:
            subschema = subschema.remove(subschema.get_field_index(col))

    md_list = []
    for keys, subgroup in data_df.groupby(partition_keys):
        if not isinstance(keys, tuple):
            keys = (keys,)