[Data] Deflake test_from_pandas_refs_e2e by removing ordering assumptions

[Partth101]

This PR addresses intermittent test failures in the execution optimizer integration suite. The test test_from_pandas_refs_e2e previously assumed a deterministic row ordering when reading from multiple pandas references, which is not guaranteed by the Ray Data interface in a distributed execution environment.

Changes
Replaced manual tuple-list comparisons with the rows_same utility from ray.data._internal.util.

Refactored assertions to use ds.to_pandas() for robust, order-agnostic data validation.

Applied the fix to all three affected assertions within the test function.

Fixes 60553

[gemini-code-assist]

Code Review

This pull request effectively addresses the flakiness in test_from_pandas_refs_e2e by removing the assumption of deterministic row ordering. The switch to using the rows_same utility for order-agnostic comparison is a solid approach.

I've added one suggestion to cache the concatenated DataFrame to improve readability and avoid redundant computation, which mirrors the pattern in the original test code.

Overall, the changes are correct and improve test stability.

[gemini-code-assist]

To improve readability and avoid re-computing the concatenated DataFrame, you can store pd.concat([df1, df2]) in a variable and reuse it for both assertions. This also aligns with the pattern in the original test code where the expected rows were computed once.

Suggested change

	df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
	ds = ray.data.from_pandas_refs([ray.put(df1), ray.put(df2)])
	values = [(r["one"], r["two"]) for r in ds.take(6)]
	rows = [(r.one, r.two) for _, r in pd.concat([df1, df2]).iterrows()]
	assert values == rows
	assert rows_same(ds.to_pandas(), pd.concat([df1, df2]))
	# Check that metadata fetch is included in stats.
	assert "FromPandas" in ds.stats()
	assert ds._plan._logical_plan.dag.name == "FromPandas"
	# Test chaining multiple operations
	ds2 = ds.map_batches(lambda x: x)
	values = [(r["one"], r["two"]) for r in ds2.take(6)]
	assert values == rows
	assert rows_same(ds2.to_pandas(), pd.concat([df1, df2]))
	df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
	expected_df = pd.concat([df1, df2])
	ds = ray.data.from_pandas_refs([ray.put(df1), ray.put(df2)])
	assert rows_same(ds.to_pandas(), expected_df)
	# Check that metadata fetch is included in stats.
	assert "FromPandas" in ds.stats()
	assert ds._plan._logical_plan.dag.name == "FromPandas"
	# Test chaining multiple operations
	ds2 = ds.map_batches(lambda x: x)
	assert rows_same(ds2.to_pandas(), expected_df)

[Partth101]

I have implemented this suggestion

[Partth101]

@bveeramani Can I get a review on this please?

Related Issue: 60553

[bveeramani]

Nice