Class Operator

python/pathway/internals/operator.py:84–191

Abstraction over operator node. Operator holds its inputs (all arguments passed to operator) and outputs (resulting tables). Inputs and outputs retain their original order.

class Operator(ABC):
    """Abstraction over operator node.

    Operator holds its inputs (all arguments passed to operator) and outputs (resulting tables).
    Inputs and outputs retain their original order.
    """

    _inputs: dict[str, InputHandle]
    _outputs: dict[str, OutputHandle]
    trace: Trace
    graph: SetOnceProperty[ParseGraph] = SetOnceProperty()
    id: int
    error_log: tables.Table | None

    def __init__(self, id: int) -> None:
        self.id = id
        self._inputs = {}
        self._outputs = {}
        self.trace = Trace.from_traceback()

    @property
    def output_tables(self) -> Iterable[tables.Table]:
        return (output.value for output in self.outputs)

    @cached_property
    def intermediate_and_output_tables(self) -> Iterable[tables.Table]:
        return list(
            chain(
                *(
                    table._id_column.context.intermediate_tables()
                    for table in self.output_tables
                ),
                self.output_tables,
            )
        )

    @property
    def input_tables(self) -> StableSet[tables.Table]:
        return StableSet.union(
            *(i.value._operator_dependencies() for i in self._inputs.values())
        )

    @property
    def inputs(self) -> list[InputHandle]:
        return list(self._inputs.values())

    @property
    def outputs(self) -> list[OutputHandle]:
        return list(self._outputs.values())

    def get_input(self, name: str) -> InputHandle:
        return self._inputs[name]

    def get_output(self, name: str) -> OutputHandle:
        return self._outputs[name]

    def get_table(self, name: str) -> tables.Table:
        return self._outputs[name].value