[data] Fix bug in memory reservation

[raulchen]

Why are these changes needed?

Fix bugs in ReservationOpResourceAllocator (introduced by #43171)

• We treat a map op and its following non-map ops as the same group. update_resources already handles this properly . But _should_unblock_streaming_output_backpressure and _op_outputs_reserved_remaining didn't consider this.
• Since we don't reserve any resources for limit and streaming_split, should set num_cpus=0 for their tasks.
• _reserved_for_op_outputs currently also includes op's internal output buffers. This is incorrect, because when preserve_order=True, task outputs will accumulate in op's internal output buffer, and use all the memory budget from _reserved_for_op_outputs. Then we still don't have memory budget to pull the blocks from the internal output buffer. Excluding the internal output buffer from _reserved_for_op_outputs fixes this issue.

Also deflake test_backpressure_from_output and test_e2e_autoscaling_up, as they depend on physical memory size of the node.

Related issue number

closes #43493
closes #43490

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 :(

[c21]

LG

[c21]

limi2 typo

[bveeramani]

Nit: To test all existing combinations with insert_limit_op, it might be cleaner to do another parametrize:

@pytest.mark.parametrize(
    "cluster_cpus", "cluster_obj_store_mem_mb",
    [
        (3, 500),
        (4, 100),
        (3, 100),
    ]
)
@pytest.mark.parametrize("insert_limit_op", [False, True])

[can-anyscale]

Can you help check that the failing tests on premerge are same as ones failing. in go/flaky. Thankks

[raulchen]

@can-anyscale checked. one is an existing flaky test, the other is because hugging face is down now.