[data] write_delta implementation

[abhishekverma-ray]

Supports SaveMode.APPEND and SaveMode.OVERWRITE (a prototype scope, by design): workers write Parquet files via pyarrow.dataset.write_dataset and report AddAction metadata; the driver commits all files in one Delta transaction via the deltalake library, creating the table if it doesn't exist. Also supports partition_by (Hive-style), storage_options, and optional name/description table metadata.

Adds a new test suite covering what this implementation actually supports: APPEND/OVERWRITE round-trips, unsupported-mode rejection, single- and multi-column partitioning, multi-block writes, table name/description metadata, the empty-dataset error path, and storage_options passthrough to the commit calls.

[gemini-code-assist]

Code Review

This pull request introduces a prototype implementation for writing Ray Datasets to Delta Lake tables via a new DeltaDatasink and a corresponding Dataset.write_delta method, supporting both append and overwrite modes. The review feedback highlights three key improvement opportunities: first, ensuring that storage_options are correctly propagated to the PyArrow FileSystem on workers to prevent write failures on secured cloud storage; second, optimizing empty APPEND operations to return early as a no-op instead of raising errors or committing empty transactions; and third, robustifying the partition parsing logic to correctly handle URL-encoded keys and map PyArrow's default null partition string (__HIVE_DEFAULT_PARTITION__) to None.

[goutamvenkat-anyscale]

Can we make this human driven comments? Verified empirically for example is not needed. Let's make this succinct.

[goutamvenkat-anyscale]

Need to write each block to a separate parquet file, collect schemas, and then on the driver unify_schemas

[abhishekverma-ray]

Sure, will change this to write out the blocks without accumulating them (which was being done to unify schemas within worker - this part will be removed). unify_schemas at driver is happening currently which would remain the same.

[abhishekverma-ray]

Done in d0affef. write() now writes each non-empty block straight to its own Parquet file and returns one schema per block; DeltaWriteResult.schema became schemas: List[pa.Schema] and the driver unifies across all workers.

I checked this doesn't regress the earlier bot findings about uncast files, by committing deliberately mismatched files directly: a null column against a string one, and a block missing a column another has, both read back correctly ([None, 'x']) with only the unified schema in the log — Delta reconciles per-file schemas on read. The one case the cast nominally covered, int32 vs int64, can't occur: unify_schemas rejects it before any cast would run.

[goutamvenkat-anyscale]

Don't need this function, just call pa.unify_schemas on the driver.

[abhishekverma-ray]

use existing unify_schemas() from transform_pyarrow.py instead of this helper function.

[abhishekverma-ray]

helper function _resolve_schema removed.

[goutamvenkat-anyscale]

Remove comment

[abhishekverma-ray]

Removed

[goutamvenkat-anyscale]

Let's use transform_pyarrow's unify_schemas instead

[abhishekverma-ray]

Switched both call sites over. Worth noting for anyone touching this later: it has to stay at the default promote_types=False. At the default it still raises ArrowTypeError for int64-vs-string and re-raises pyarrow's original exception, so the except pa.ArrowTypeError incompatibility check is unaffected. With promote_types=True pyarrow switches to promote_options="permissive" and silently coerces string+int64 → string, which would quietly disable that check.

[goutamvenkat-anyscale]

Just compare against https://delta-io.github.io/delta-rs/api/delta_writer/#deltalake.write_deltalake . If it's functionally equivalent, please use it. If not, then fine.

[abhishekverma-ray]

write_deltalake(data=...) only accepts in-memory Arrow: it normalizes the input to a RecordBatchReader and hands it to Rust write() / write_to_deltalake(), both typed data: RecordBatchReader. It writes the Parquet itself in the calling process and never references AddAction — there's no parameter anywhere for committing already-written files. Since our workers write their own Parquet in parallel and send back only metadata, adopting it would mean routing every worker's data through the driver, or doing per-worker commits and losing the single atomic commit. create_table_with_add_actions / create_write_transaction are the only metadata-only commit APIs in the package, which is what we already use.

[goutamvenkat-anyscale]

Look at partitioning.py to see if the parser already exists.

[abhishekverma-ray]

Looked — PathPartitionParser._parse_hive_path does exist and does work on a bare relative path with no filesystem access. Three gaps stop it being a drop-in here:

It unquotes values but not keys.
No HIVE_DEFAULT_PARTITION → None mapping; it returns the literal string, which would break the null-partition round-trip. That convention is Hive/Delta-specific and appears nowhere in partitioning.py.
It returns only the keys present in the path, with no {col: None} backfill for every requested partition_by column — and with field_names set it raises on a count/name mismatch rather than filling in.

[cursor]

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

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