[Refactor] [Data]: test_limit_pushdown_conservative - spilt tests & fix ordering assumption

[ryankert01]

Description

Split multi-case test function

• test_limit_pushdown_conservative → 10 separate tests (basic fusion, limit fusion reversed, multiple limit fusion, maprows, mapbatches, filter, project, sort, complex interweaved operations, and between two map operators)

Fixed ordering assumptions

• Added check_ordering=False to union tests (blocks may interleave)
• Added check_ordering=False to project test with override_num_blocks (parallel execution)

Related issues

Related to #58655

Additional information

[gemini-code-assist]

Code Review

This pull request is a great refactoring effort, splitting a large, multi-case test function into ten smaller, more focused tests. This significantly improves the readability and maintainability of the test suite. The fixes for ordering assumptions by adding check_ordering=False are also correct and well-justified for parallel execution and union operations. I've added a few suggestions to further enhance the refactoring by replacing helper identity functions with more concise lambda expressions.

[gemini-code-assist]

For conciseness and better readability, the f1 identity function can be replaced with an inline lambda x: x. This change also requires updating the expected plan string to use Map(<lambda>) instead of Map(f1).

Suggested change

	def f1(x):
	return x
	ds = ray.data.range(100, override_num_blocks=100).map(f1).limit(1)
	_check_valid_plan_and_result(
	ds, "Read[ReadRange] -> Limit[limit=1] -> MapRows[Map(f1)]", [{"id": 0}]
	)
	ds = ray.data.range(100, override_num_blocks=100).map(lambda x: x).limit(1)
	_check_valid_plan_and_result(
	ds, "Read[ReadRange] -> Limit[limit=1] -> MapRows[Map(<lambda>)]", [{"id": 0}]
	)

[gemini-code-assist]

Similar to other tests, the f2 identity function can be replaced with a lambda x: x for better code conciseness. The expected plan string should be updated accordingly.

    ds = ray.data.range(100, override_num_blocks=100).map_batches(lambda x: x).limit(1)
    _check_valid_plan_and_result(
        ds,
        "Read[ReadRange] -> Limit[limit=1] -> MapBatches[MapBatches(<lambda>)]",
        [{"id": 0}],
    )

[ryankert01]

PTAL @owenowenisme @bveeramani

[bveeramani]

Except for the tests listed below, the Ray Data pipelines in this diff aren't guaranteed to produce rows in a specific output order. So, if you don't set check_ordering=False, the tests might raise false negatives.

Could you update the appropriate tests to use check_ordering=False?

List of tests with guaranteed order (okay to use check_ordering=True):

• test_limit_pushdown_union_with_sort
• test_limit_pushdown_complex_interweaved_operations
• test_limit_pushdown_stops_at_sort

[cursor]

Bug: Missing check_ordering parameter in parametrized test (Bugbot Rules)

The test test_limit_pushdown_udf_modifying_row_count_with_map_batches calls _check_valid_plan_and_result without check_ordering=False, but the pipeline doesn't include a .sort() operation to guarantee row order. According to the reviewer's feedback, all tests without guaranteed order need check_ordering=False to avoid false negatives from non-deterministic row ordering in Ray Data pipelines.

python/ray/data/tests/test_execution_optimizer_limit_pushdown.py#L594-L599

https://github.com/ray-project/ray/blob/c17fca7a4eef301dca5ac4b3c3fa1006ab522a87/python/ray/data/tests/test_execution_optimizer_limit_pushdown.py#L594-L599

 

---

[ryankert01]

@bveeramani Sorry for the late response. I walked through the code but still wasn’t able to conclude safely. I’ve gone ahead and updated them with check_ordering=False.

[bveeramani]

LGTM!

[bveeramani]

Just enabled auto-merge. ty for the contribution!