[Data] Avoid merging map ops in cases when it leads to substantial parallelism reduction

[alexeykudinkin]

Why are these changes needed?

These changes are needed to prevent cases when Map Operator fusion could lead to substantial parallelism reduction.

Consider following scenario:

1. Upstream MapOp specifies min_rows_per_input_bundle=1
2. Downstreeam MapOp specifies min_rows_per_input_bundle=100

For a dataset of 100 rows and 10 blocks (10 rows / block) if we do fuse in this case, fused operator's parallelism will be just 1 task (determined by downstream) substantially reducing upstream's parallelism.

This is a big issue when we fuse Read ops with subsequent Map operations.

This change:

1. Avoids fusion of Read ops with downstream Map ops that have batch_size specified
2. Adjusts fusion sequence to avoid fusing operators with substantial reduction in estimated parallelism
3. Adds telemetry

Related issue number

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

[raulchen]

In addition to read ops, I think we should disable fusion as long as the previous map op doesn't preserve num rows (e.g., read, filter, map_batches, flat_map, etc)

[raulchen]

for map_batches, typically it preserves num rows.
But today we don't enforce that.
related issue #36295
One option is to enforce that by default, and add a flag to allow violation.

[alexeykudinkin]

One option is to enforce that by default, and add a flag to allow violation.

I don't think we can do that anymore with our public API -- i can totally see that being too limiting.

Regardless, though preserving num-rows for proper limit push-downs is an important topic but tangential to this change.