[Data] Add Zarr Datasource (duplicate PR of 63003)

[ArturNiederfahrenhorst]

Description

Keeping some changes to the #63003 in this feature branch.

[ArturNiederfahrenhorst]

These are all taken from #63003.
I assume that they are useful in matching patterns @alexandrplashchinsky

[ArturNiederfahrenhorst]

We use this call_with_retry pattern in iceberg, lance and parquet so I assume that it's worth propagating the pattern to this PR.

[ArturNiederfahrenhorst]

This produces arrays of different dimensions so we expect Ray Data to handle pyarrow's variable shaped tensors downstream. We could also force tensors of the same length here and use padding.

[ArturNiederfahrenhorst]

These reads are sequential and if we read from some web interface that is super slow, this make become a significant I/O bottleneck. A simple solution would be a threadpool to remove the IO bottleneck.

[ArturNiederfahrenhorst]

This case happened to me with some UMI dataset

[ArturNiederfahrenhorst]

These are skipped but the dataset exists.
Just skipping because all other datasources in ray data also skip.
If we didn't skip, the docsbuild would require the zarr python package.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

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

[cursor]

Sequential I/O reads cause significant remote-store bottleneck

Medium Severity

In _create_aligned_read_fn, all chunk reads across all arrays and all batch items are done sequentially in a single thread. For N aligned arrays and M descriptors in a batch, this performs N×M sequential I/O operations. Similarly in _create_read_fn, all batch chunks are read sequentially. When reading from remote stores (S3, GCS, HTTPS), these sequential reads form a significant I/O bottleneck since each read involves a full network round-trip with no concurrency.

Additional Locations (1)

• python/ray/data/_internal/datasource/zarrv2_datasource.py#L330-L341

 

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

[cursor]

chunk_shape validation allows zero and negative values

Low Severity

The chunk_shape validation only checks that the value is a list or tuple but doesn't verify elements are positive integers. Passing chunk_shape=[0] would later cause a ZeroDivisionError inside grid_shape (via math.ceil(s / c) where c=0), producing a confusing traceback far from the validation site. The error message promises "non-empty sequence of positive integers" but the check doesn't enforce this.

 

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

[ArturNiederfahrenhorst]

Closing in favor of #63003