[BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp

[laraschmidt]

A DoFn may emit elements with a timestamp up to DoFn#getAllowedTimestampSkew() before the current element's timestamp. This change implements this change for timer's as well. Now a timer may have an output timestamp up to DoFn#getAllowedTimestampSkew() before the current element's timestamp. Before this change a timer's output timestamp could not be before the current output element.

Additional Context: https://lists.apache.org/thread.html/r7554658114ddde86c5d82e1c39fe7e1ef587fe926b8e406d1130d501%40%3Cdev.beam.apache.org%3E

@reuvenlax

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) 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.

See the Contributor Guide for more tips on how to make review process smoother.

ValidatesRunner compliance status (on master branch)

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---					---
Java
Python	---					---
XLang						---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[lukecwik]

Is there an associated JIRA?

[reuvenlax]

I think you have to update the similar check in FnApiDoFnRunner

[laraschmidt]

Added checks in FnApiDoFnRunner which was not checking for skew on timers and not checking timestamps at all for normal emit. PTAL and make sure it makes sense and I can prob add some tests in FnApiRunner.

PTAL @reuvenlax @lukecwik Thanks!

[lukecwik]

Won't this allow for infinite skew since if have a timer at X and skew of -1 then the first time the timer is processed you can output at time X-1 and when it gets scheduled again you can now output at X-2 since the the new timers timestamp is X-1?

The changes to the FnApiDoFnRunner to check timestamp output validity makes sense.

[lukecwik]

You can handle the proper bounds via:

Instant lowerBound;
try {
   lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
catch (ArithmeticException e) {
   lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (outputTimestamp.isBefore(lowerBound)) {
  ...
}

Finally it would make sense to check the upper bound as well of BoundedWindow.TIMESTAMP_MAX_VALUE

[laraschmidt]

Sounds good. Done.

[lukecwik]

Can you not use TimerMap as it limits the number of runners this can run on and use individual timers?

[laraschmidt]

The elements need to share a timer then but I suppose the same logic will hit whenever we set the output timestamp. I removed the output of elements though since this would be harder. Wasn't really needed anyway.

[lukecwik]

Need to tag with UsesTimersInParDo.class and/or UsesTimerMap.class.

[laraschmidt]

Done. Assuming just the validates runner tests need this because I don't see it elsewhere in the file. Let me know if not.

[laraschmidt]

Won't this allow for infinite skew since if have a timer at X and skew of -1 then the first time the timer is processed you can output at time X-1 and when it gets scheduled again you can now output at X-2 since the the new timers timestamp is X-1?

So my understanding of the reason for these checks is to stop people from doing the wrong thing without realizing it. We don't even take any different action based on this variable. It seems okay to apply this to each specific output timestamp and let you skew more if you chain timers in this fashion.

On a more practical note, there's reasons why you might want a timer to output an earlier element if you've properly set up watermark holds. There's currently no way to do that so we need some allowance. It would probably be better if we could constrain skew from the first output timestamp but I don't think that's available in the later timers, right?

If you disagree with the approach, I can bring this up on the email thread for others to chime in in case they are not checking here.

[laraschmidt]

PTAL, @lukecwik @reuvenlax

[je-ik]

+1
Do we have a follow-up for un-deprececating the method?

[lukecwik]

Won't this allow for infinite skew since if have a timer at X and skew of -1 then the first time the timer is processed you can output at time X-1 and when it gets scheduled again you can now output at X-2 since the the new timers timestamp is X-1?

So my understanding of the reason for these checks is to stop people from doing the wrong thing without realizing it. We don't even take any different action based on this variable. It seems okay to apply this to each specific output timestamp and let you skew more if you chain timers in this fashion.

On a more practical note, there's reasons why you might want a timer to output an earlier element if you've properly set up watermark holds. There's currently no way to do that so we need some allowance. It would probably be better if we could constrain skew from the first output timestamp but I don't think that's available in the later timers, right?

If you disagree with the approach, I can bring this up on the email thread for others to chime in in case they are not checking here.

I think users will be surprised that their data will be dropped as late once they pass the watermark skew bound if they output past it. The existing logic had guards for this explicitly since it would be surprising for users so I do believe it is important enough to discuss whether there is another approach to solve this or we are ok with this happening.

[laraschmidt]

Won't this allow for infinite skew since if have a timer at X and skew of -1 then the first time the timer is processed you can output at time X-1 and when it gets scheduled again you can now output at X-2 since the the new timers timestamp is X-1?

So my understanding of the reason for these checks is to stop people from doing the wrong thing without realizing it. We don't even take any different action based on this variable. It seems okay to apply this to each specific output timestamp and let you skew more if you chain timers in this fashion.
On a more practical note, there's reasons why you might want a timer to output an earlier element if you've properly set up watermark holds. There's currently no way to do that so we need some allowance. It would probably be better if we could constrain skew from the first output timestamp but I don't think that's available in the later timers, right?
If you disagree with the approach, I can bring this up on the email thread for others to chime in in case they are not checking here.

I think users will be surprised that their data will be dropped as late once they pass the watermark skew bound if they output past it. The existing logic had guards for this explicitly since it would be surprising for users so I do believe it is important enough to discuss whether there is another approach to solve this or we are ok with this happening.

We chatted a bit about this offline. There's actually no guarantee that the watermark is held back when using DoFn#getAllowedTimestampSkew. The allowedTimestampSkew just removes the check that we have to avoid accidentally dropping late data. See the javadoc [1] and relevant reply from Jan [2].

[1] https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/transforms/DoFn.html#getAllowedTimestampSkew--
[2] https://lists.apache.org/thread.html/r34c70d8a5f213f7bd2f4557019e27b7f07f5120d0a8794512c88568c%40%3Cdev.beam.apache.org%3E

[laraschmidt]

PTAL @lukecwik

[lukecwik]

This is looking great, need to fix up the OnWindowExpiration case and add a ValidatesRunner test for it.

[codecov]

Codecov Report

Merging #15540 (7b0de72) into master (bc82f6f) will increase coverage by 37.42%.
The diff coverage is n/a.

❗ Current head 7b0de72 differs from pull request most recent head c43415e. Consider uploading reports for the commit c43415e to get more accurate results

@@             Coverage Diff             @@
##           master   #15540       +/-   ##
===========================================
+ Coverage   46.10%   83.53%   +37.42%     
===========================================
  Files         196      445      +249     
  Lines       19498    61210    +41712     
===========================================
+ Hits         8990    51131    +42141     
- Misses       9538    10079      +541     
+ Partials      970        0      -970     

Impacted Files	Coverage Δ
sdks/go/pkg/beam/io/textio/sdf.go
sdks/go/pkg/beam/testing/teststream/teststream.go
sdks/go/pkg/beam/core/util/reflectx/util.go
...ks/go/pkg/beam/core/runtime/coderx/coderx.shims.go
sdks/go/pkg/beam/core/runtime/exec/emit.go
sdks/go/pkg/beam/core/runtime/genx/genx.go
sdks/go/pkg/beam/validate.go
sdks/go/pkg/beam/core/graph/coder/bytes.go
sdks/go/pkg/beam/transforms/stats/sum_switch.go
sdks/go/pkg/beam/testing/passert/count.go
... and 631 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update bc82f6f...c43415e. Read the comment docs.

[laraschmidt]

@lukecwik PTAL, still need to check that everything runs internally but otherwise should be good.

[lukecwik]

Looks great and I really like the simplification in the testing.

[lukecwik]

Run Java PreCommit