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

Code before:
    def _rewire_output_deps(
        self,
        source_op: "PhysicalOperator",
        input_dependencies: List["PhysicalOperator"],
    ) -> None:
        for input_op in input_dependencies:
            assert isinstance(input_op, PhysicalOperator), input_op
            input_op._output_dependencies = [
                dep for dep in input_op._output_dependencies if dep is not source_op
            ]
            if self not in input_op._output_dependencies:
                input_op._output_dependencies.append(self)

    def _wire_output_deps(self, input_dependencies: List["PhysicalOperator"]) -> None:
        for input_op in input_dependencies:
            assert isinstance(input_op, PhysicalOperator), input_op
            input_op._output_dependencies.append(self)

    def set_logical_operators(
        self,
        *logical_ops: LogicalOperator,
    ):
        self._logical_operators = list(logical_ops)

    @property
    def target_max_block_size_override(self) -> Optional[int]:
        """
        Target max block size output by this operator. If this returns None,
        then the default from DataContext should be used.
        """
        if self._output_block_size_option_override is None:
            return None
        else:
            return self._output_block_size_option_override.target_max_block_size

    def override_target_max_block_size(self, target_max_block_size: Optional[int]):
        self._output_block_size_option_override = OutputBlockSizeOption.of(
            target_max_block_size=target_max_block_size
        )

    def mark_execution_finished(self):
        """Manually mark that this operator has finished execution."""
        self._is_execution_marked_finished = True

    def has_execution_finished(self) -> bool:
        """Return True when this operator has finished execution.

        The outputs may or may not have been taken.
        """
        from ..operators.base_physical_operator import InternalQueueOperatorMixin


Review comment (refactor):
`_wire_output_deps` can introduce `duplicate` downstream entries when multiple inputs are replaced by the same node.

### Example

#### Initial DAG (two inputs):
```
A -> R
B -> R
```
• `R` has two upstreams: `A` and `B`
• Reverse edges:
```
A.output_deps = [R]
B.output_deps = [R]
```

#### Transform: replace both A and B with the same node X
```
def transform(op):
    if op is A or op is B:
        return X   # same object
    return op
```

After transform:
```
R.input_dependencies = [X, X]
```

#### What `_wire_output_deps` does

- It appends `R` once per input:
```
for input_op in input_dependencies:
    input_op._output_dependencies.append(self)
```

- So we get:
```
X.output_deps = [R, R]
```

**Notice**: `R` appears twice.

#### Why this is a problem
Many downstream checks rely on len(output_dependencies):
- **Resource accounting** might think X has 2 consumers
- **Backpressure logic** might miscompute downstream capacity
-  **Completion logic** might process the same downstream twice

#### Expected behavior
Even if both inputs point to the same node, logically `R` is still only one downstream:
```
X.output_deps = [R]
```

Code after:
        # intentionally raises. Use a side-effect-free shallow copy to avoid
        # re-running __init__ wiring/ID/metrics initialization during transform.
        target = object.__new__(type(self))
        target.__dict__ = self.__dict__.copy()
        # The transformed node should have a distinct identity and metrics owner.
        target._id = str(uuid.uuid4())
        target._metrics = OpRuntimeMetrics(target)
        # The copied node belongs to a new transformed DAG. Reverse edges are
        # rewired by parents, so avoid carrying stale downstream references.
        target._output_dependencies = []
        return target

    def _rewire_output_deps(
        self,
        source_op: "PhysicalOperator",
        input_dependencies: List["PhysicalOperator"],
    ) -> None:
        for input_op in input_dependencies:
            assert isinstance(input_op, PhysicalOperator), input_op
            input_op._output_dependencies = [
                dep for dep in input_op._output_dependencies if dep is not source_op
            ]
            if self not in input_op._output_dependencies:
                input_op._output_dependencies.append(self)

    def _wire_output_deps(self, input_dependencies: List["PhysicalOperator"]) -> None:
        for input_op in input_dependencies:
            assert isinstance(input_op, PhysicalOperator), input_op
            input_op._output_dependencies.append(self)

    def set_logical_operators(
        self,
        *logical_ops: LogicalOperator,
    ):
        self._logical_operators = list(logical_ops)

    @property
    def target_max_block_size_override(self) -> Optional[int]:
        """
        Target max block size output by this operator. If this returns None,
        then the default from DataContext should be used.
        """
        if self._output_block_size_option_override is None:
            return None
        else:
            return self._output_block_size_option_override.target_max_block_size

    def override_target_max_block_size(self, target_max_block_size: Optional[int]):
        self._output_block_size_option_override = OutputBlockSizeOption.of(
            target_max_block_size=target_max_block_size
        )