[Data] Disable throttling for UnionOperator

[slfan1989]

Description

This PR disables resource-based throttling for UnionOperator by overriding the throttling_disabled() method to return True.

Why it's needed:

• UnionOperator only manipulates bundle metadata (routes RefBundle objects between input/output buffers)
• It doesn't launch any Ray tasks or consume computational resources (CPU/memory)
• Currently it gets allocated resources unnecessarily, which is wasteful and impacts scheduling efficiency
• This aligns UnionOperator with other metadata-only operators (LimitOperator, OutputSplitter, AggregateNumRows) that already have throttling disabled

Related issues

Link related issues: "Fixes #60627", "Closes #60627", or "Related to #60627".

Additional information

Implementation details:

Added throttling_disabled() method in python/ray/data/_internal/execution/operators/union_operator.py:

def throttling_disabled(self) -> bool:
    """Disables resource-based throttling.

    Union operator only manipulates bundle metadata and doesn't launch
    any tasks, so it doesn't need resource allocation.
    """
    return True

Testing:
Added two unit tests in python/ray/data/tests/test_operators.py:

1. test_union_operator_throttling_disabled() - Verifies that throttling_disabled() returns True
2. test_union_operator_throttling_matches_similar_operators() - Ensures consistency with LimitOperator and OutputSplitter

[gemini-code-assist]

Code Review

This pull request correctly disables resource-based throttling for the UnionOperator by overriding the throttling_disabled() method. This is a good optimization, as the operator only manipulates metadata and does not consume computational resources. The change is accompanied by two new unit tests that effectively verify the new behavior and ensure consistency with other similar operators. The implementation is clean and the reasoning is sound. I have one minor suggestion to simplify one of the new tests to improve its clarity and focus.

[gemini-code-assist]

The setup for this test can be simplified. Since throttling_disabled() is a static property of the operator and doesn't depend on its inputs, we can instantiate UnionOperator without any input operators. This makes the test more focused and easier to read.

Suggested change

	ctx = DataContext.get_current()
	# Create input buffers with some test data
	input_bundles_1 = make_ref_bundles([[1, 2, 3]])
	input_bundles_2 = make_ref_bundles([[4, 5, 6]])
	input_op1 = InputDataBuffer(ctx, input_bundles_1)
	input_op2 = InputDataBuffer(ctx, input_bundles_2)
	# Create union operator
	union_op = UnionOperator(ctx, input_op1, input_op2)
	ctx = DataContext.get_current()
	union_op = UnionOperator(ctx)