[Data] Add path_column support to ParquetDatasource

[daiping8]

Motivation

This change addresses a limitation in Ray's Parquet reading functionality where the file path column name is hardcoded to "path". This causes data loss when reading files that already contain a "path" column, as the original column values are overwritten with file paths.

The fix enables users to:

• Avoid column name conflicts by specifying a custom name
• Use more descriptive column names for their use cases
• Maintain backward compatibility with existing code

Related Issue

Closed #63757

Implementation Details

Core API Changes

1. python/ray/data/read_api.py

   • Added path_column parameter to read_parquet() function signature
   • Updated documentation to describe the new parameter
   • Handled V2 datasource path column logic in column selection

2. python/ray/data/datasource/file_based_datasource.py

   • Added path_column parameter to __init__
   • Stored as instance variable with default "path"
   • Updated fill_column call to use custom name

Key Design Decisions

1. Default behavior preserved: When path_column=None, defaults to "path" for full backward compatibility

2. Consistent across versions: Both V1 and V2 datasources support the feature identically

3. Schema inference aware: The custom column name is respected during schema inference, preventing duplicate columns

4. Minimal API surface: Only exposes path_column at the read_parquet() API level; internal threading is transparent

Verification

Unit Tests

1. V1 datasource test:

   pytest python/ray/data/tests/datasource/test_parquet.py::test_include_paths_with_custom_column -v

2. V2 datasource test:

   pytest python/ray/data/_internal/datasource_v2/tests/test_parquet_datasource_v2.py::test_infer_schema_with_custom_path_column -v

Manual Verification

1. Test conflict avoidance:

   import ray
   import pyarrow.parquet as pq
   import pyarrow as pa
   
   # Create file with existing "path" column
   table = pa.Table.from_pydict({"animals": ["cat", "dog"], "path": ["a", "b"]})
   pq.write_table(table, "test.parquet")
   
   # Read with custom path column name
   ds = ray.data.read_parquet("test.parquet", include_paths=True, path_column="source_path")
   rows = ds.take_all()
   
   # Verify both columns exist
   assert "path" in rows[0]  # Original data column
   assert "source_path" in rows[0]  # Custom file path column
   assert rows[0]["path"] in ["a", "b"]  # Original values preserved
   assert rows[0]["source_path"] == "test.parquet"  # File path correct

2. Test backward compatibility:

   # Default behavior unchanged
   ds = ray.data.read_parquet("file.parquet", include_paths=True)
   assert "path" in ds.schema()

3. Test schema inference:

   datasource = ParquetDatasourceV2(
       ["file.parquet"], 
       include_paths=True, 
       path_column="file_path"
   )
   schema = datasource.infer_schema(manifest)
   assert "file_path" in schema.names

Integration Testing

Run the full parquet datasource test suite:

pytest python/ray/data/tests/datasource/test_parquet.py -v
pytest python/ray/data/_internal/datasource_v2/tests/test_parquet_datasource_v2.py -v

[gemini-code-assist]

Code Review

This pull request introduces a path_column parameter to read_parquet and the underlying datasources, allowing users to specify a custom column name for file paths when include_paths is enabled. This prevents conflicts with existing columns in the dataset. The review feedback suggests normalizing self._path_column to its default value ('path') during initialization in ParquetDatasource._init_state, which would simplify subsequent references by removing redundant fallback checks.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit bd22664ec707d6bc9447dd865513939b72e75926. Configure here.}

[github-actions]

This pull request has been automatically marked as stale because it has not had
any activity for 14 days. It will be closed in another 14 days if no further activity occurs.
Thank you for your contributions.

You can always ask for help on our discussion forum or Ray's public slack channel.

If you'd like to keep this open, just leave any comment, and the stale label will be removed.

[daiping8]

keep

[richardliaw]

OK just coming back to this PR. Feels like we should just introduce a string typing to include_paths similar to how Polars does this. What do you think?

read(include_paths: str | None = None) and if True, we default to path but we can raise a warning saying that the variable typestring has changed?