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 2021/02/17 05:31:21 UTC
[GitHub] [beam] kennknowles opened a new pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
kennknowles opened a new pull request #14001:
URL: https://github.com/apache/beam/pull/14001
Discovered this by running on a runner that:
1. Does not deserialize to the Java class and call `isNonMerging()` (which all the Java-based runners do)
2. Does not special-case the session window URN unless the NEEDS_MERGE bit is set.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [x] 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.
- [x] Update `CHANGES.md` with noteworthy changes.
- [x] 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] kennknowles commented on pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14001:
URL: https://github.com/apache/beam/pull/14001#issuecomment-780311079
@TheNeuralBit @y1chi I think this is related to #13998
----------------------------------------------------------------
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] y1chi commented on a change in pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14001:
URL: https://github.com/apache/beam/pull/14001#discussion_r577878578
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
##########
@@ -295,10 +299,17 @@ public static FunctionSpec toProto(WindowFn<?, ?> windowFn, SdkComponents compon
.setAllowedLateness(windowingStrategy.getAllowedLateness().getMillis())
.setTrigger(TriggerTranslation.toProto(windowingStrategy.getTrigger()))
.setWindowFn(windowFnSpec)
- .setAssignsToOneWindow(windowingStrategy.getWindowFn().assignsToOneWindow())
+ .setAssignsToOneWindow(windowFn.assignsToOneWindow())
+ .setMergeStatus(
+ windowFn instanceof InvalidWindows
+ ? (((InvalidWindows) windowFn).getOriginalWindowFn().isNonMerging()
Review comment:
probably stupid question: if the InvalidWindow only happens after GBK or combine, does the original window fn matter? should the MergeStatus always be ALREADY_MERGED for InvalidWindow? How should runner treat ALREADY_MERGED differently than NON_MERGING?
----------------------------------------------------------------
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] kennknowles commented on a change in pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #14001:
URL: https://github.com/apache/beam/pull/14001#discussion_r577900336
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
##########
@@ -295,10 +299,17 @@ public static FunctionSpec toProto(WindowFn<?, ?> windowFn, SdkComponents compon
.setAllowedLateness(windowingStrategy.getAllowedLateness().getMillis())
.setTrigger(TriggerTranslation.toProto(windowingStrategy.getTrigger()))
.setWindowFn(windowFnSpec)
- .setAssignsToOneWindow(windowingStrategy.getWindowFn().assignsToOneWindow())
+ .setAssignsToOneWindow(windowFn.assignsToOneWindow())
+ .setMergeStatus(
+ windowFn instanceof InvalidWindows
+ ? (((InvalidWindows) windowFn).getOriginalWindowFn().isNonMerging()
Review comment:
`InvalidWindows` is only used in exactly one place, only for merging `originalWindowFn`: https://github.com/apache/beam/blob/7c43ab6a8df9b23caa7321fddff9a032a71908f6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java#L216
----------------------------------------------------------------
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] chamikaramj commented on a change in pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on a change in pull request #14001:
URL: https://github.com/apache/beam/pull/14001#discussion_r581321380
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
##########
@@ -295,10 +299,17 @@ public static FunctionSpec toProto(WindowFn<?, ?> windowFn, SdkComponents compon
.setAllowedLateness(windowingStrategy.getAllowedLateness().getMillis())
.setTrigger(TriggerTranslation.toProto(windowingStrategy.getTrigger()))
.setWindowFn(windowFnSpec)
- .setAssignsToOneWindow(windowingStrategy.getWindowFn().assignsToOneWindow())
+ .setAssignsToOneWindow(windowFn.assignsToOneWindow())
+ .setMergeStatus(
+ windowFn instanceof InvalidWindows
+ ? (((InvalidWindows) windowFn).getOriginalWindowFn().isNonMerging()
Review comment:
Are we assuming that usage here (not just the InvalidWindows spec) ? If so probably this at least needs a comment about that assumption.
----------------------------------------------------------------
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] y1chi commented on a change in pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14001:
URL: https://github.com/apache/beam/pull/14001#discussion_r577878578
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
##########
@@ -295,10 +299,17 @@ public static FunctionSpec toProto(WindowFn<?, ?> windowFn, SdkComponents compon
.setAllowedLateness(windowingStrategy.getAllowedLateness().getMillis())
.setTrigger(TriggerTranslation.toProto(windowingStrategy.getTrigger()))
.setWindowFn(windowFnSpec)
- .setAssignsToOneWindow(windowingStrategy.getWindowFn().assignsToOneWindow())
+ .setAssignsToOneWindow(windowFn.assignsToOneWindow())
+ .setMergeStatus(
+ windowFn instanceof InvalidWindows
+ ? (((InvalidWindows) windowFn).getOriginalWindowFn().isNonMerging()
Review comment:
probably stupid question: if the InvalidWindow only happens after GBK or combine, does the original window fn matter? should the MergeStatus always be ALREADY_MERGED for InvalidWindow? How should runner tread ALREADY_MERGED differently than NON_MERGING?
----------------------------------------------------------------
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] kennknowles commented on pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14001:
URL: https://github.com/apache/beam/pull/14001#issuecomment-780697641
----------------------------------------------------------------
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] kennknowles commented on a change in pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #14001:
URL: https://github.com/apache/beam/pull/14001#discussion_r577899082
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
##########
@@ -295,10 +299,17 @@ public static FunctionSpec toProto(WindowFn<?, ?> windowFn, SdkComponents compon
.setAllowedLateness(windowingStrategy.getAllowedLateness().getMillis())
.setTrigger(TriggerTranslation.toProto(windowingStrategy.getTrigger()))
.setWindowFn(windowFnSpec)
- .setAssignsToOneWindow(windowingStrategy.getWindowFn().assignsToOneWindow())
+ .setAssignsToOneWindow(windowFn.assignsToOneWindow())
+ .setMergeStatus(
+ windowFn instanceof InvalidWindows
+ ? (((InvalidWindows) windowFn).getOriginalWindowFn().isNonMerging()
Review comment:
It is a good question. I think the original WindowFn is just for giving a good error message. It is also used to check compatibility of `Flatten` of two PCollections.
----------------------------------------------------------------
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] kennknowles merged pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles merged pull request #14001:
URL: https://github.com/apache/beam/pull/14001
----------------------------------------------------------------
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] kennknowles commented on pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14001:
URL: https://github.com/apache/beam/pull/14001#issuecomment-784693921
Confirmed that the new test is running so the changes to the JUnit runner didn't make things start silently passing: https://scans.gradle.com/s/mqlyl7yrzs5be/tests?anchor=:runners:core-construction-java-test-org.apache.beam.runners.core.construction-windowingstrategytranslationtest$expectedprototests-testsessionsmergestatus-882&search=WindowingStrategyTranslation
----------------------------------------------------------------
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] kennknowles commented on a change in pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #14001:
URL: https://github.com/apache/beam/pull/14001#discussion_r581395321
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowingStrategyTranslation.java
##########
@@ -295,10 +299,17 @@ public static FunctionSpec toProto(WindowFn<?, ?> windowFn, SdkComponents compon
.setAllowedLateness(windowingStrategy.getAllowedLateness().getMillis())
.setTrigger(TriggerTranslation.toProto(windowingStrategy.getTrigger()))
.setWindowFn(windowFnSpec)
- .setAssignsToOneWindow(windowingStrategy.getWindowFn().assignsToOneWindow())
+ .setAssignsToOneWindow(windowFn.assignsToOneWindow())
+ .setMergeStatus(
+ windowFn instanceof InvalidWindows
+ ? (((InvalidWindows) windowFn).getOriginalWindowFn().isNonMerging()
Review comment:
Per dev@ list discussion, I'm going to change this to not special-case `InvalidWindows`. For now: if it is a merging WindowFn it says NEEDS_MERGE, otherwise it doesn't.
Once I finish adjusting the SDKs to actually track ALREADY_MERGED bit, I will adjust the translation too.
----------------------------------------------------------------
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] kennknowles commented on pull request #14001: [BEAM-11824] Set merge status in WindowingStrategyTranslation
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14001:
URL: https://github.com/apache/beam/pull/14001#issuecomment-780315144
Yet another fun remnant of the earliest days of windowing and triggering. It looks like in the design fo the proto, we had an idea for an improved way of expressing this. Maybe Python already has done this differently than Java?
Since `InvalidWindows` already cannot be used, it would also be acceptably backwards-compatible to replace it with different non-crashing behavior. I don't know.
----------------------------------------------------------------
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