[Data] Don't push limit past map_batches by default

[bveeramani]

In Ray 2.52, we made a breaking change that allowed limits to incorrectly get pushed down past map_batches that produce more output rows than input rows received. This can happen for example if you're using map_batches to filter or download data.

This PR reverts the breaking change to the (correct) behavior of Ray <2.52.

[gemini-code-assist]

Code Review

This pull request changes the default value of udf_modifying_row_count in map_batches to True, making the API safer by default by disabling limit pushdown for map_batches unless explicitly enabled. This is a great improvement for correctness. The docstring update is clear, and the new test correctly validates the new behavior.

One side effect is a minor performance regression for internal methods like add_column and drop_columns that use map_batches without modifying row counts. These will now default to the safer, non-optimized path. A follow-up to explicitly set udf_modifying_row_count=False for these specific calls would be beneficial to restore optimal performance.

Overall, the change is a solid improvement to the API's robustness.

[cursor]

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

[alexeykudinkin]

This is a breaking change

[bveeramani]

This is a breaking change

This is a breaking change from 2.52 and 2.53, but it's both the correct behavior and the behavior that's consistent from <2.52. We made a breaking change between 2.51 and 2.52

[xinyuangui2]

Why do we need to manually set udf_modifying_row_count=False here

[bveeramani]

Because of this logic here:

ray/python/ray/data/_internal/logical/rules/operator_fusion.py

Lines 652 to 666 in e32a2f2

	# Do not fuse Map operators in case:
	#
	# - Upstream could (potentially) drastically modify number of rows, while
	# - Downstream has `min_rows_per_input_bundle` specified
	#
	# Fusing such transformations is not desirable as it could
	#
	# - Drastically reduce parallelism for the upstream up (for ex, if
	# fusing ``Read->MapBatches(batch_size=...)`` with large enough batch-size
	# could drastically reduce parallelism level of the Read op)
	#
	# - Potentially violate batching semantic by fusing
	# ``Filter->MapBatches(batch_size=...)``
	#
	if (

.

By default, we don't know if a map_batches will modify the number of rows. The user needs to explicitly set udf_modifying_row_count

[alexeykudinkin]

We can't break fusion like that. We'd have relax conditional you're referencing above

[alexeykudinkin]

Discussed offline:

• We'll be fixing fusion in the follow-up