[FEAT] Overwrite mode for write parquet/csv

[colin-ho]

Addresses: #3112 and #1768

Implements overwrite mode for write_parquet and write_csv.

Upon finishing the write, we are left with a manifest of written file paths. We can use this to perform a delete all files not in manifest, by:

1. Do an ls to figure out all the current files in the root dir.
2. Use daft's built in is_in expression to get the file paths to delete.
3. Delete them.

Notes:

• Relies on fsspec for ls and rm functionalities. This is favored over pyarrow filesystem because rm is a bulk delete method, aka we can do the delete in a single API call. Pyarrow filesystem does not have bulk deletes.

[codspeed-hq]

CodSpeed Performance Report

Merging #3108 will not alter performance

_{Comparing colin/overwrite-writes (5fdc9d9) with main (5b450fb)}

Summary

✅ 17 untouched benchmarks

[codecov]

Codecov Report

Attention: Patch coverage is 87.50000% with 3 lines in your changes missing coverage. Please review.

Project coverage is 78.63%. Comparing base (c69ee3f) to head (5fdc9d9).
Report is 41 commits behind head on main.

Files with missing lines	Patch %	Lines
daft/dataframe/dataframe.py	77.77%	2 Missing ⚠️
daft/filesystem.py	93.33%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3108      +/-   ##
==========================================
- Coverage   78.65%   78.63%   -0.02%     
==========================================
  Files         618      621       +3     
  Lines       73192    74150     +958     
==========================================
+ Hits        57568    58310     +742     
- Misses      15624    15840     +216     

Files with missing lines	Coverage Δ
daft/filesystem.py	70.62% <93.33%> (+1.90%)	⬆️
daft/dataframe/dataframe.py	86.48% <77.77%> (-0.09%)	⬇️

... and 34 files with indirect coverage changes

[jaychia]

Can we use Union[Literal["append"], Literal["overwrite"]]?

[jaychia]

I would avoid fsspec if possible here so we can avoid taking a dependency on it.

I'm also not sure if the bulk delete API has any performance benefit over a serial delete call... I guess for the services (e.g. S3) we can parallelize DELETE requests over the wire.

Ideally we can support DELETE on our own IO clients, but in the absence of that shall we use PyArrow instead and just naively delete one-by-one?

[colin-ho]

Yeah we can use PyArrow, should be easier to implement anyways since we already have the machinery to infer a pyarrow filesystem. I can also run some tests to see if paralellizing the deletes make sense.

[samster25]

It doesn't look like we're supporting partition overwrite mode here. Where we only delete files if we wrote a new file into that partition.

Could we leverage what @desmondcheongzx is working on for hive style reads to be able to do this?

[colin-ho]

Probably, but we could also just match on the directory paths and delete accordingly. Will make a separate PR for this.