[Data] Convert one-to-one logical operators to frozen dataclasses

[myandpr]

Description

This PR implements converting one-to-one logical operators to frozen dataclasses.

Why this is needed:

• This is the first operator-group step for the frozen logical-operator migration under [Data] Make LogicalPlan stateless and comparable #60312.
• It removes in-place mutation paths for one-to-one logical operators.
• It keeps the change scoped to logical-layer behavior needed for D1.

What this PR changes:

• Converts one-to-one logical operators to frozen dataclasses:
  • Limit
  • Download
• Applies one-to-one construction cleanup for frozen compatibility:
  • uses InitVar[LogicalOperator] + __post_init__ to initialize _name, _input_dependencies, and _num_outputs
  • makes Download.ray_remote_args a canonical dict field (default_factory=dict)
• Adds frozen-safe transform behavior:
  • Limit._apply_transform() recreates Limit when input changes
  • Download._apply_transform() recreates Download when input changes
• Updates optimizer rules to avoid mutating frozen instances:
  • limit_pushdown.py: recreate Limit/Download on input replacement
  • predicate_pushdown.py: recreate Limit on input replacement
• Adds regression coverage for Limit(Download(...)) under limit pushdown.
• Scope is intentionally D1-only (one-to-one logical operators); no map/all-to-all or physical-layer changes in this PR.

Related issues

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

Additional information

Tests

Added/updated:

• python/ray/data/tests/test_execution_optimizer_limit_pushdown.py
  • adds regression for Limit(Download(...)) under limit pushdown with frozen operators

Validated with targeted existing tests:

• python/ray/data/tests/test_execution_optimizer_limit_pushdown.py
• python/ray/data/tests/test_predicate_pushdown.py
• python/ray/data/tests/test_operator_fusion.py
• python/ray/data/tests/test_execution_optimizer_basic.py
• python/ray/data/tests/test_execution_optimizer_advanced.py
• python/ray/data/tests/test_projection_fusion.py
• python/ray/data/tests/test_randomize_block_order.py
• python/ray/data/tests/test_state_export.py::test_logical_op_args
• python/ray/data/tests/unit/test_logical_plan.py

Stack Plan

To complete #60312, the original stack was:

1. [Data] Use input_dependencies property in logical operator #60529
2. [Data][2/4] Rename logical operator attributes to public #60528
3. [Data] Drop output_dependencies from logical operators #60530
4. [Data] Convert logical operators to frozen dataclasses #60531

Since PR4 was still too large, it is being further split:

1. PR-A: default LogicalOperator naming behavior [Data] Default LogicalOperator name to class name #61020
2. PR-B: move output_dependencies responsibility to physical side [Data] Move output_dependencies responsibilities to PhysicalOperator #61107
3. PR-C: make LogicalOperator an ABC with abstract num_outputs [Data] Make LogicalOperator an ABC with abstract num_outputs #61308
4. PR-D: convert logical operators to frozen dataclasses in small groups (D1/D2/D3)
   • D1: one-to-one operators （this PR）
   • D2: map operators
   • D3: all-to-all + join/read/write groups (as needed)

Planned follow-ups (not blocking this stack):

• Converting all input_op usage to input_dependencies
• Potential AbstractFrom restructuring

[gemini-code-assist]

Code Review

This pull request refactors the Limit and Download logical operators to be frozen dataclasses, which is a great step towards ensuring the immutability of the logical plan and improving overall robustness. The necessary adjustments in the optimizer rules (limit_pushdown, predicate_pushdown) are correctly implemented, and a new test is added to cover these changes. I have a couple of suggestions to make the optimizer rules more robust and maintainable in light of these changes.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

[myandpr]

Heads-up: the current microcheck failure appears unrelated to this PR's D1 changes. The failing test is python/ray/data/tests/test_read_datasource.py::test_read_datasource_compute_strategy, which is asserting on read_op._compute (private field), while Read exposes compute as a public attribute. This maybe came from #59633 and can be fixed in a small follow-up PR by switching _compute -> compute in that test.

I have submit #61423 to fix this issue.

[bveeramani]

What happens if we default to the LogicalOperator._apply_transform implementation and don't override this?

[myandpr]

We can’t use the default _apply_transform for frozen Limit/Download yet: Operator._apply_transform does copy.copy(self) and then assigns _input_dependencies, which raises FrozenInstanceError on frozen dataclasses. That’s why these one-to-one operators override _apply_transform and rebuild nodes instead of mutating fields.

[bveeramani]

When LogicalOperator is a pure abstract class (no constructor parameters), I think we should work towards avoiding __setattr__. Maybe it could look like this:

num_outputs: Optional[int] = field(init=False, default=None, repr=False)

@property
def name(self) -> str:
    return f"limit={self.limit"}"

@property
def input_dependencies(self):
    return [self.input_op]

Okay as-is because this is a transient state and I trust it'll all look good in the end

[bveeramani]

Same question here

[myandpr]

Same root cause as above: default _apply_transform does copy.copy + input_dependencies reassignment, which is invalid for frozen dataclasses; this override keeps Download frozen-safe by rebuilding on input change.

[bveeramani]

What's the motivation for this change/how does it relate to this refactor? What happens if we remove it?

[myandpr]

For predicate_pushdown, the Limit branch is for the same reason: the generic clone path uses copy.copy(op) + input_dependencies reassignment. Removing the Limit special-case would reintroduce FrozenInstanceError when pushing predicates through Limit.