gemini-code-assist[bot] commented on code in PR #38268: URL: https://github.com/apache/beam/pull/38268#discussion_r3583944613
########## sdks/python/apache_beam/pipeline.py: ########## @@ -112,6 +112,42 @@ __all__ = ['Pipeline', 'transform_annotations'] +def _descendant_applied_ptransforms( + current: 'AppliedPTransform') -> set['AppliedPTransform']: + descendants = set() + pending = list(current.parts) + while pending: + part = pending.pop() + if part in descendants: + continue + descendants.add(part) + pending.extend(part.parts) + return descendants + + +def _register_top_level_side_outputs( + current: 'AppliedPTransform', result: pvalue.PCollection) -> None: + if not result._side_outputs: + return + + descendants = _descendant_applied_ptransforms(current) + for side_tag, side_pcoll in result._side_outputs.items(): Review Comment:  Accessing `result._side_outputs` directly can raise an `AttributeError` if the `PCollection` was reconstructed (e.g., via `from_runner_api` or other deserialization paths) where `_side_outputs` is not populated in the instance's `__dict__`. Using `getattr(result, '_side_outputs', None)` is safer and prevents potential crashes during pipeline replacement or other graph operations on reconstructed pipelines. ```python side_outputs = getattr(result, '_side_outputs', None) if not side_outputs: return descendants = _descendant_applied_ptransforms(current) for side_tag, side_pcoll in side_outputs.items(): ``` ########## sdks/python/apache_beam/pvalue.py: ########## @@ -138,19 +139,105 @@ def __or__(self, ptransform): return self.pipeline.apply(ptransform, self) +class _SideOutputsContainer: + """Lightweight accessor over named side-output PCollections. + + Supports attribute access (``container.dropped``), indexing + (``container["dropped"]``), iteration over tag names, ``len()``, and + ``in``. ``__getattr__`` on a missing tag raises ``AttributeError`` with + a message listing available tags. + """ + def __init__(self, side_outputs: dict[str, 'PCollection']): + object.__setattr__(self, '_side_outputs', dict(side_outputs)) + + def __getattr__(self, tag: str) -> 'PCollection': + if tag.startswith('__'): + raise AttributeError(tag) + try: + return self._side_outputs[tag] + except KeyError as exc: + available = sorted(self._side_outputs) + raise AttributeError( + f"No side output named {tag!r}. Available: {available}") from exc + + def __getitem__(self, tag: str) -> 'PCollection': + return self._side_outputs[tag] + + def __iter__(self): + return iter(self._side_outputs) + + def __len__(self): + return len(self._side_outputs) + + def __contains__(self, tag): + return tag in self._side_outputs + + class PCollection(PValue, Generic[T]): """A multiple values (potentially huge) container. Dataflow users should not construct PCollection objects directly in their pipelines. """ + def __init__( + self, + pipeline: 'Pipeline', + tag: Optional[str] = None, + element_type: Optional[Union[type, 'typehints.TypeConstraint']] = None, + windowing: Optional['Windowing'] = None, + is_bounded=True): + super().__init__( + pipeline, + tag=tag, + element_type=element_type, + windowing=windowing, + is_bounded=is_bounded) + self._side_outputs: Optional[dict[str, 'PCollection']] = None + def __eq__(self, other): if isinstance(other, PCollection): return self.tag == other.tag and self.producer == other.producer def __hash__(self): return hash((self.tag, self.producer)) + def __copy__(self): + result = type(self).__new__(type(self)) + result.__dict__ = self.__dict__.copy() + return result + + @property + def side_outputs(self) -> _SideOutputsContainer: + return _SideOutputsContainer(self._side_outputs or {}) Review Comment:  Accessing `self._side_outputs` directly in the `side_outputs` property will raise an `AttributeError` if the `PCollection` was reconstructed (e.g., from the runner API) or created without calling `__init__` (which is common in some serialization or mocking frameworks). Using `getattr(self, '_side_outputs', None)` ensures that this property safely returns an empty container instead of crashing. ```suggestion return _SideOutputsContainer(getattr(self, '_side_outputs', None) or {}) ``` ########## sdks/python/apache_beam/pvalue.py: ########## @@ -138,19 +139,105 @@ def __or__(self, ptransform): return self.pipeline.apply(ptransform, self) +class _SideOutputsContainer: + """Lightweight accessor over named side-output PCollections. + + Supports attribute access (``container.dropped``), indexing + (``container["dropped"]``), iteration over tag names, ``len()``, and + ``in``. ``__getattr__`` on a missing tag raises ``AttributeError`` with + a message listing available tags. + """ + def __init__(self, side_outputs: dict[str, 'PCollection']): + object.__setattr__(self, '_side_outputs', dict(side_outputs)) + + def __getattr__(self, tag: str) -> 'PCollection': + if tag.startswith('__'): + raise AttributeError(tag) Review Comment:  If `_side_outputs` is not yet initialized or is missing from `__dict__` (for example, during unpickling or copying), accessing `self._side_outputs` inside `__getattr__` will trigger `__getattr__('_side_outputs')`, leading to infinite recursion and a `RecursionError`. Restricting dynamic attribute lookup to exclude any attributes starting with an underscore (or specifically `_side_outputs`) prevents this infinite recursion. ```suggestion if tag.startswith('_'): raise AttributeError(tag) ``` ########## sdks/python/apache_beam/pvalue.py: ########## @@ -138,19 +139,105 @@ def __or__(self, ptransform): return self.pipeline.apply(ptransform, self) +class _SideOutputsContainer: + """Lightweight accessor over named side-output PCollections. + + Supports attribute access (``container.dropped``), indexing + (``container["dropped"]``), iteration over tag names, ``len()``, and + ``in``. ``__getattr__`` on a missing tag raises ``AttributeError`` with + a message listing available tags. + """ + def __init__(self, side_outputs: dict[str, 'PCollection']): + object.__setattr__(self, '_side_outputs', dict(side_outputs)) Review Comment:  Since `_SideOutputsContainer` does not override `__setattr__`, using `object.__setattr__` is unnecessary and can be confusing. Standard attribute assignment `self._side_outputs = dict(side_outputs)` is cleaner and more idiomatic. ```suggestion def __init__(self, side_outputs: dict[str, 'PCollection']): self._side_outputs = dict(side_outputs) ``` ########## sdks/python/apache_beam/pvalue.py: ########## @@ -138,19 +139,105 @@ def __or__(self, ptransform): return self.pipeline.apply(ptransform, self) +class _SideOutputsContainer: + """Lightweight accessor over named side-output PCollections. + + Supports attribute access (``container.dropped``), indexing + (``container["dropped"]``), iteration over tag names, ``len()``, and + ``in``. ``__getattr__`` on a missing tag raises ``AttributeError`` with + a message listing available tags. + """ + def __init__(self, side_outputs: dict[str, 'PCollection']): + object.__setattr__(self, '_side_outputs', dict(side_outputs)) + + def __getattr__(self, tag: str) -> 'PCollection': + if tag.startswith('__'): + raise AttributeError(tag) + try: + return self._side_outputs[tag] + except KeyError as exc: + available = sorted(self._side_outputs) + raise AttributeError( + f"No side output named {tag!r}. Available: {available}") from exc + + def __getitem__(self, tag: str) -> 'PCollection': + return self._side_outputs[tag] + + def __iter__(self): + return iter(self._side_outputs) + + def __len__(self): + return len(self._side_outputs) + + def __contains__(self, tag): + return tag in self._side_outputs Review Comment:  Implementing `__dir__` on `_SideOutputsContainer` allows interactive environments (like Jupyter notebooks or IPython) to auto-complete the available side output tags, greatly improving usability. ```suggestion def __contains__(self, tag): return tag in self._side_outputs def __dir__(self): return sorted(super().__dir__() + list(getattr(self, '_side_outputs', {}).keys())) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
