Repository: ray-project/ray
File: python/ray/data/_internal/logical/interfaces/logical_operator.py (Python)

Code before:
from abc import ABC, abstractmethod
from enum import Enum
from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Optional

from .operator import Operator
from ray.data.block import BlockMetadata
from ray.data.expressions import Expr

if TYPE_CHECKING:
    from ray.data.block import Schema


class LogicalOperator(Operator):
    """Abstract class for logical operators.

    A logical operator describes transformation, and later is converted into
    physical operator.
    """

    def __init__(
        self,
        name: Optional[str],
        input_dependencies: List["LogicalOperator"],
        num_outputs: Optional[int] = None,
    ):
        super().__init__(
            name,
            input_dependencies,
        )
        for x in input_dependencies:
            assert isinstance(x, LogicalOperator), x

        self.num_outputs: Optional[int] = num_outputs

    @property
    def name(self) -> str:
        name = super().name
        if name is None:
            return self.__class__.__name__
        return name

    def estimated_num_outputs(self) -> Optional[int]:
        """Returns the estimated number of blocks that
        would be outputted by this logical operator.

Review comment (refactor):
In either this PR or a follow-up, could we simplify all the logical operators that explicitly default to the class name?

For example, I think we can update `Count`:
```diff
+        super().__init__([input_op])
-        super().__init__("Count", [input_op])
```
https://github.com/ray-project/ray/blob/c3f787b5d8871f3ffabcb0b33529df5417d2cbed/python/ray/data/_internal/logical/operators/count_operator.py#L22

I think this should be a mostly mechanical change that I can stamp, and it will make it easier to review the later PRs 

Code after:
from abc import ABC, abstractmethod
from enum import Enum
from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Optional

from .operator import Operator
from ray.data.block import BlockMetadata
from ray.data.expressions import Expr

if TYPE_CHECKING:
    from ray.data.block import Schema


class LogicalOperator(Operator):
    """Abstract class for logical operators.

    A logical operator describes transformation, and later is converted into
    physical operator.
    """

    def __init__(
        self,
        input_dependencies: List["LogicalOperator"],
        num_outputs: Optional[int] = None,
        *,
        name: Optional[str] = None,
    ):
        if name is None:
            name = self.__class__.__name__
        super().__init__(
            name,
            input_dependencies,
        )
        for x in input_dependencies:
            assert isinstance(x, LogicalOperator), x

        self.num_outputs: Optional[int] = num_outputs

    def estimated_num_outputs(self) -> Optional[int]:
        """Returns the estimated number of blocks that
        would be outputted by this logical operator.

        This method does not execute the plan, so it does not take into consideration
        block splitting. This method only considers high-level block constraints like
        `Dataset.repartition(num_blocks=X)`. A more accurate estimation can be given by