compression not passed to engine.write_partition

[birdsarah]

I'm trying to debug a separate issue (my files never start writing) and so I've been carefully reading parquet/core.py

If I'm understanding it correctly, the compression keyword is not passed to engine.write_partition

https://github.com/dask/dask/blob/master/dask/dataframe/io/parquet/core.py#L458-L481

    # write parts
    dwrite = delayed(engine.write_partition)
    parts = [
        dwrite(
            d,
            path,
            fs,
            filename,
            partition_on,
            write_metadata_file,
            fmd=meta,
            index_cols=index_cols,
            **kwargs_pass
        )
        for d, filename in zip(df.to_delayed(), filenames)
    ]

    # single task to complete
    out = delayed(lambda x: None)(parts)
    if write_metadata_file:
        out = delayed(engine.write_metadata)(
            parts, meta, fs, path, append=append, compression=compression
        )

kwargs_pass will not have compression in as it's a _to_parquet_kwargs

cc @rjzamora

[rjzamora]

Looks to me like you are correct @birdsarah - This seems like a bug. I will try to explore this soonish.

[jmcotelo]

I opened #5266 a while ago and it seems this issue is the very same thing. I had no time to properly sit down and write tests for it. In any case, if @birdsarah solves this, #5266 could be closed as well.