[Data] Make zip operator accept multiple input

[owenowenisme]

Why are these changes needed?

Before making zip operator a streaming operator, we make it accept multiple input first.

Now Zip operator can be used with

>>> import ray
>>> ds1 = ray.data.range(5)
>>> ds2 = ray.data.range(5)
>>> ds3 = ray.data.range(5)
>>> ds1.zip(ds2, ds3).take_batch()
{'id': array([0, 1, 2, 3, 4]), 'id_1': array([0, 1, 2, 3, 4]), 'id_2': array([0, 1, 2, 3, 4])}                                                                         
>>> ds1.zip(ds2, ds3).take_all()
[{'id': 0, 'id_1': 0, 'id_2': 0}, {'id': 1, 'id_1': 1, 'id_2': 1}, {'id': 2, 'id_1': 2, 'id_2': 2}, {'id': 3, 'id_1': 3, 'id_2': 3}, {'id': 4, 'id_1': 4, 'id_2': 4}]
>>>   

Related issue number

#56504

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[owenowenisme]

@richardliaw @gvspraveen @alexeykudinkin PTAL, thanks!

[goutamvenkat-anyscale]

Thanks for your contribution! Overall the first change looks good. Just a few minor comments.

[iamjustinhsu]

typically when there are changes to the baseline, we would want to fix it. Is this possible to fix or is this a bug with pydoc linting?

[owenowenisme]

Yeah I am aware of this, but I thought this is intended? Other operator does not have data_context in its doc
e.g. UnionOperator

    def __init__(
        self,
        data_context: DataContext,
        *input_ops: PhysicalOperator,
    ):
        """Create a UnionOperator.

        Args:
            input_ops: Operators generating input data for this operator to union.
        """

[richardliaw]

ah in the future we can just add data_context into the doc, i think that's a good thing to fix

[alexeykudinkin]

This should be min, not max

[alexeykudinkin]

Let's make sure this is covered with tests also

[owenowenisme]

Thanks! On second thought, neither max nor min seems accurate, right? Since the number of input blocks for each output should be the same (to perform a zip), and we already assert:

        total_left_rows = sum(left_block_rows)
        total_right_rows = sum(right_block_rows)
        if total_left_rows != total_right_rows:
            raise ValueError(
                "Cannot zip datasets of different number of rows: "
                f"{total_left_rows}, {total_right_rows}"
            )

Maybe we don't actually need to calculate num_outputs here?
Correct me if I'm wrong, thanks!

[owenowenisme]

Even if we use ‎min for the number of output rows right now, this logic will need to change when user-directed dropping or padding is introduced.

Padding would require using ‎max, while dropping would use ‎min, so the calculating number of rows here is redundant.

Discussed with @gvspraveen offline.

[alexeykudinkin]

Same comment

[alexeykudinkin]

Same commetn

[cursor]

Bug: Index Assertion Error Causes Buffer Access Failure

The assertion for input_index in _add_input_inner allows an out-of-bounds index equal to len(self._input_dependencies). This can lead to an IndexError when accessing self._input_buffers.