You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/12/14 22:05:18 UTC
[GitHub] [beam] yifanmai opened a new pull request #13549: [BEAM-11196] Fix `None` parent when fusing >2 stages
yifanmai opened a new pull request #13549:
URL: https://github.com/apache/beam/pull/13549
Fixes a bug introduced by #13202 where the parent for a fused stage (or packed combiner) gets set to `None` if more than two stages are fused (or packed). This occurs because `_lowest_common_ancestor` uses the `subtransforms` field of the `PipelineContext`'s transforms to figure out parent-child relationships. However, a stage that is created by fusion or by another optimizer phase will not be present in the `PipelineContext`'s transforms. As a result, such a stage is passed to `_lowest_common_ancestor`, it will return `None`. This change fixes it by additionally using the `parent` field of the input stages.
This bug only affects runners that use `translations.pipeline_from_stages` with `partial=True`.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb merged pull request #13549: [BEAM-11196] Fix `None` parent when fusing >2 stages
Posted by GitBox <gi...@apache.org>.
robertwb merged pull request #13549:
URL: https://github.com/apache/beam/pull/13549
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a change in pull request #13549: [BEAM-11196] Fix `None` parent when fusing >2 stages
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #13549:
URL: https://github.com/apache/beam/pull/13549#discussion_r542921512
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -1217,23 +1215,33 @@ def get_ancestors(name):
def _parent_for_fused_stages(stages, context):
- # type: (Iterable[Optional[str]], TransformContext) -> Optional[str]
+ # type: (Iterable[Stage], TransformContext) -> Optional[str]
'''Returns the name of the new parent for the fused stages.
The new parent is the lowest common ancestor of the fused stages that is not
contained in the set of stages to be fused. The provided context is used to
compute ancestors of stages.
'''
+
+ parents = context.parents_map()
+ # If any of the input stages were produced by fusion or an optimizer phase,
Review comment:
Ack. Thanks for the clarification.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] yifanmai commented on pull request #13549: [BEAM-11196] Fix `None` parent when fusing >2 stages
Posted by GitBox <gi...@apache.org>.
yifanmai commented on pull request #13549:
URL: https://github.com/apache/beam/pull/13549#issuecomment-744741720
R: @robertwb @tvalentyn @pabloem
Could we prioritize getting this fix into the release cut, since we're also enabling combiner packing in #13455? Apologies for last minute-ness!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] yifanmai commented on a change in pull request #13549: [BEAM-11196] Fix `None` parent when fusing >2 stages
Posted by GitBox <gi...@apache.org>.
yifanmai commented on a change in pull request #13549:
URL: https://github.com/apache/beam/pull/13549#discussion_r542919666
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -1217,23 +1215,33 @@ def get_ancestors(name):
def _parent_for_fused_stages(stages, context):
- # type: (Iterable[Optional[str]], TransformContext) -> Optional[str]
+ # type: (Iterable[Stage], TransformContext) -> Optional[str]
'''Returns the name of the new parent for the fused stages.
The new parent is the lowest common ancestor of the fused stages that is not
contained in the set of stages to be fused. The provided context is used to
compute ancestors of stages.
'''
+
+ parents = context.parents_map()
+ # If any of the input stages were produced by fusion or an optimizer phase,
Review comment:
Nope-ish. This problem really arises because for stages going through the phases, there are two sources of truth for their parents: `parent` field of the stage and the `subtransforms` field of transforms in the `PipelineContext`. The parent of a stage _has_ to be a transform in PipelineContext, which means there is only one source of truth (the `subtransforms` field). So we don't encounter this issue with grandparents.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a change in pull request #13549: [BEAM-11196] Fix `None` parent when fusing >2 stages
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #13549:
URL: https://github.com/apache/beam/pull/13549#discussion_r542916880
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -1217,23 +1215,33 @@ def get_ancestors(name):
def _parent_for_fused_stages(stages, context):
- # type: (Iterable[Optional[str]], TransformContext) -> Optional[str]
+ # type: (Iterable[Stage], TransformContext) -> Optional[str]
'''Returns the name of the new parent for the fused stages.
The new parent is the lowest common ancestor of the fused stages that is not
contained in the set of stages to be fused. The provided context is used to
compute ancestors of stages.
'''
+
+ parents = context.parents_map()
+ # If any of the input stages were produced by fusion or an optimizer phase,
Review comment:
Is there a case where the parent's parent is not correctly set?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org