[Data] Make Zip a properly streaming operator

[Hyunoh-Yeo]

Description

Reimplements ZipOperator as a streaming operator. Previously, zip bulk-materialized all blocks from both inputs into the object store before producing any output, with throttling_disabled() returning True and the operator listed as a blocking-materializing operator. This reworks it to align and zip blocks incrementally as they arrive.

How alignment works: whenever a block is available from every input, the operator takes one from each, splits any larger blocks down to the minimum row count (carrying the remainder forward as a per-input "leftover"), zips the aligned blocks, and emits the output bundle immediately. Mismatched total row counts are detected at all_inputs_done() and raise ValueError.

Related issues

Closes #56300

Additional information

[gemini-code-assist]

Code Review

This pull request refactors the ZipOperator from a bulk implementation to a streaming implementation, allowing it to process blocks incrementally and align rows by splitting blocks as needed. It also enables throttling for the operator and adds comprehensive tests. The review feedback highlights critical performance concerns regarding synchronous ray.get calls inside _try_output() and _fill_block_deque(), which block the main streaming executor thread and can lead to scheduling bottlenecks or deadlocks.

[Hyunoh-Yeo]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors ZipOperator from a bulk operator to a fully streaming operator. Blocks are now processed incrementally as they arrive from all inputs, aligned dynamically using a split-and-leftover mechanism, and zipped asynchronously via DataOpTasks. It also introduces asynchronous row-count resolution when metadata is missing and enables backpressure support. Since there are no review comments, I have no feedback to provide.

[400Ping]

Please resolve conflicts

[owenowenisme]

Hey Hyunoh! A few high level design questions:

1. Can we drop the physical split task? _dispatch_ready_zips fires a separate _split_block_at_row remote task per misaligned boundary just to slice a block, then feeds the head into a second zip task. Since slicing is offset-based and cheap, could we skip the split entirely, carry the leftover as (block_ref, offset, num_rows) and let the zip task slice each input down to min_rows locally before zipping?Kinda similar the streaming-repartition approach. It removes one remote task + object-store round-trip per boundary and collapses split+zip into a single task.
2. _validate_if_settled feels heavier than it needs to be. And why do we need all these different deque & list(_staging, _block_deques, leftovers)?
3. For the output queue I think we need to use ReorderingBundleQueue when preseve_order is set.

[cursor]

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

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

[cursor]

Count tasks skip hold tracking

Medium Severity

Row-count fetches never take a _SourceBundle hold, and clear_internal_input_queue still drops the pending slice that was keeping that bundle alive. On early finish, for example a downstream limit, that can dequeue and destroy_if_owned a block while its MetadataOpTask is still running, and the count callback is never cancelled.

Additional Locations (1)

• python/ray/data/_internal/execution/operators/zip_operator.py#L314-L333

 

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