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/24 04:45:38 UTC
[GitHub] [beam] kennknowles opened a new pull request #14059: Remove InvalidWindows from Java SDK, instead track "already merged" bit
kennknowles opened a new pull request #14059:
URL: https://github.com/apache/beam/pull/14059
Per dev@ list discussion, after merge, a merging windowing strategy becomes inert, just like non-merging WindowFn.
------------------------
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] kennknowles merged pull request #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles merged pull request #14059:
URL: https://github.com/apache/beam/pull/14059
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14059:
URL: https://github.com/apache/beam/pull/14059#discussion_r583302403
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
##########
@@ -332,7 +332,7 @@
public WindowingStrategy<?, ?> getOutputStrategyInternal(WindowingStrategy<?, ?> inputStrategy) {
WindowingStrategy<?, ?> result = inputStrategy;
if (getWindowFn() != null) {
- result = result.withWindowFn(getWindowFn());
+ result = result.withAlreadyMerged(false).withWindowFn(getWindowFn());
Review comment:
I see, thanks for the explanation.
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14059:
URL: https://github.com/apache/beam/pull/14059#issuecomment-786989184
Fixing commit history before merge. Flink & Dataflow both passed VR and they are the runners touched by this.
----------------------------------------------------------------
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 #14059: Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14059:
URL: https://github.com/apache/beam/pull/14059#issuecomment-785436905
Run Java PreCommit
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14059:
URL: https://github.com/apache/beam/pull/14059#issuecomment-786823516
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14059:
URL: https://github.com/apache/beam/pull/14059#discussion_r583297512
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
##########
@@ -332,7 +332,7 @@
public WindowingStrategy<?, ?> getOutputStrategyInternal(WindowingStrategy<?, ?> inputStrategy) {
WindowingStrategy<?, ?> result = inputStrategy;
if (getWindowFn() != null) {
- result = result.withWindowFn(getWindowFn());
+ result = result.withAlreadyMerged(false).withWindowFn(getWindowFn());
Review comment:
I'm not sure if we can safely assume getWindowFn() is always non-null. Wouldn't it be safer to set already merged directly like WindowingStrategy<?, ?> result = inputStrategy.withAlreadyMerged(false)?
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #14059:
URL: https://github.com/apache/beam/pull/14059#discussion_r583300306
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
##########
@@ -332,7 +332,7 @@
public WindowingStrategy<?, ?> getOutputStrategyInternal(WindowingStrategy<?, ?> inputStrategy) {
WindowingStrategy<?, ?> result = inputStrategy;
if (getWindowFn() != null) {
- result = result.withWindowFn(getWindowFn());
+ result = result.withAlreadyMerged(false).withWindowFn(getWindowFn());
Review comment:
This is a "model" question, sort of:
- if the user calls `Window.into(WindoFn)` we definitely reset `alreadyMerged`.
- if the user calls `Window.remerge()` we reset `alreadyMerged`.
- if the user calls `Window.configure()` to set a new trigger, accumulation mode, or allowed lateness: should we reset `alreadyMerged`? I decided that we should not. So if windowFn is null, we do not reset.
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #14059:
URL: https://github.com/apache/beam/pull/14059#discussion_r583236807
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java
##########
@@ -25,26 +25,19 @@
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.IncompatibleWindowException;
-import org.apache.beam.sdk.transforms.windowing.InvalidWindows;
import org.apache.beam.sdk.transforms.windowing.NonMergingWindowFn;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
import org.apache.beam.sdk.transforms.windowing.WindowMappingFn;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.WindowingStrategy;
import org.joda.time.Instant;
/**
* A {@link WindowFn} that leaves all associations between elements and windows unchanged.
*
* <p>This {@link WindowFn} is applied when elements must be passed through a {@link GroupByKey},
- * but should maintain their existing {@link Window} assignments. Because windows may have been
- * merged, the earlier {@link WindowFn} may not appropriately maintain the existing window
- * assignments. For example, if the earlier {@link WindowFn} merges windows, after a {@link
- * GroupByKey} the {@link WindowingStrategy} uses {@link InvalidWindows}, and no further {@link
- * GroupByKey} can be applied without applying a new {@link WindowFn}. This {@link WindowFn} allows
- * existing window assignments to be maintained across a single group by key, at which point the
- * earlier {@link WindowingStrategy} should be restored.
+ * but should maintain their existing {@link Window} assignments. This will prevent merging if the
Review comment:
This reminds me that you caught a bug: even if the user does use `Window.into(merging windowfn)` this will use `Window.configure()` that will keep the already merged bit. I will update the PR to reset the "already merged" bit whenever the WindowFn changes.
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14059:
URL: https://github.com/apache/beam/pull/14059#issuecomment-786989268
(all runners are touched by `ReduceFnRunner` but in the same way)
----------------------------------------------------------------
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 #14059: Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14059:
URL: https://github.com/apache/beam/pull/14059#issuecomment-785439022
R: @y1chi
----------------------------------------------------------------
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 #14059: [BEAM-11866] Remove InvalidWindows from Java SDK, instead track "already merged" bit
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14059:
URL: https://github.com/apache/beam/pull/14059#discussion_r583069716
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/IdentityWindowFn.java
##########
@@ -25,26 +25,19 @@
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.IncompatibleWindowException;
-import org.apache.beam.sdk.transforms.windowing.InvalidWindows;
import org.apache.beam.sdk.transforms.windowing.NonMergingWindowFn;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
import org.apache.beam.sdk.transforms.windowing.WindowMappingFn;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.WindowingStrategy;
import org.joda.time.Instant;
/**
* A {@link WindowFn} that leaves all associations between elements and windows unchanged.
*
* <p>This {@link WindowFn} is applied when elements must be passed through a {@link GroupByKey},
- * but should maintain their existing {@link Window} assignments. Because windows may have been
- * merged, the earlier {@link WindowFn} may not appropriately maintain the existing window
- * assignments. For example, if the earlier {@link WindowFn} merges windows, after a {@link
- * GroupByKey} the {@link WindowingStrategy} uses {@link InvalidWindows}, and no further {@link
- * GroupByKey} can be applied without applying a new {@link WindowFn}. This {@link WindowFn} allows
- * existing window assignments to be maintained across a single group by key, at which point the
- * earlier {@link WindowingStrategy} should be restored.
+ * but should maintain their existing {@link Window} assignments. This will prevent merging if the
Review comment:
The WindowingStrategy's alreadyMerged field will be set to true after {@link GroupByKey} and this will prevent...
----------------------------------------------------------------
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