[FEAT] Include file paths as column from read_parquet/csv/json

[colin-ho]

Addresses: #2808

This PR enables adding file path as a column from file reads via the file_path_column: str | None parameter. This works by appending a column of the file path to the Table post read + pushdowns.

Having it as a string makes it easy to have unique field name guarantees, i.e. if the user specifies a column name that already exists then an error is thrown.

[codspeed-hq]

CodSpeed Performance Report

Merging #2953 will not alter performance

_{Comparing colin/include-path-in-read (fd67611) with main (ab1b772)}

Summary

✅ 17 untouched benchmarks

[codecov]

Codecov Report

Attention: Patch coverage is 84.90566% with 24 lines in your changes missing coverage. Please review.

Project coverage is 78.50%. Comparing base (ab1b772) to head (fd67611).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
src/daft-plan/src/builder.rs	69.44%	11 Missing ⚠️
src/daft-scan/src/lib.rs	70.58%	10 Missing ⚠️
src/daft-micropartition/src/python.rs	50.00%	1 Missing ⚠️
src/daft-scan/src/glob.rs	98.52%	1 Missing ⚠️
src/daft-scan/src/python.rs	87.50%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2953      +/-   ##
==========================================
+ Coverage   78.47%   78.50%   +0.02%     
==========================================
  Files         610      610              
  Lines       71746    71865     +119     
==========================================
+ Hits        56303    56415     +112     
- Misses      15443    15450       +7     

Files with missing lines	Coverage Δ
daft/io/_csv.py	95.65% <ø> (ø)
daft/io/_json.py	91.30% <ø> (ø)
daft/io/_parquet.py	86.20% <ø> (ø)
daft/io/common.py	85.00% <ø> (ø)
src/daft-csv/src/read.rs	99.33% <ø> (ø)
src/daft-micropartition/src/micropartition.rs	90.83% <100.00%> (+0.02%)	⬆️
src/daft-micropartition/src/ops/cast_to_schema.rs	100.00% <100.00%> (ø)
src/daft-scan/src/anonymous.rs	76.92% <100.00%> (+1.24%)	⬆️
src/daft-scan/src/scan_task_iters.rs	96.95% <100.00%> (+0.01%)	⬆️
src/daft-sql/src/table_provider/read_parquet.rs	96.42% <100.00%> (+0.13%)	⬆️
... and 5 more

... and 8 files with indirect coverage changes

[kindkindk]

This is a good feature. To read file user will add df.read_parquet("...", filename="True"), note the string "True" and reason is explained in top comment.

Can we instead have fileName as boolean in Python API and then Rust can treat it as string ?

[colin-ho]

This is a good feature. To read file user will add df.read_parquet("...", filename="True"), note the string "True" and reason is explained in top comment.

Can we instead have fileName as boolean in Python API and then Rust can treat it as string ?

Thanks for the comment! and sorry for late reply.

We could have it as a boolean in the Python API as you mentioned, but I don't quite understand what you mean by having Rust treat it as string. If it was a boolean, then on the Rust side we would simply create a column called "paths" for the file paths. The issue arises when the data already has a column called "paths".

In which case i can think of two solutions:

1. retry with different variations like "_paths", "file_paths", etc
2. expose two arguments (include_file_path: bool = False and file_path_column_name: str | None = None), in which we would use the provided file_path_column_name otherwise default to "paths", and subsequently if there is an existing column with that name, throw an error and tell the user to pick a different name. This solution is essentially a more explicit variant of the current implementation, which is just the single file_path_column_name: str | None = None argument.

cc @jaychia for thoughts

[desmondcheongzx]

Two comments but overall looks good!

[desmondcheongzx]

I think here we should either add a TODO or add file_path_column to the ParquetScanBuilder, then add the argument to the SQL side as well in read_parquet.rs.