[BEAM-3304] Go triggering support #15239
Conversation
riteshghorse
changed the title
|
R: @lostluck |
There was a problem hiding this comment.
I think you're missing a few files, or it's out of date? This a good start for validating and getting a test working though. As written I can't see how you'd get to a pane encoding error however, hence guessing files are missing.
Feel free to add a new test and pipeline functions to the windowinto.go files, in order to test out triggers. It will certainly fail on the Go direct runner, so also add a filter to prevent it from running on the direct runner during the integration tests here:
| // TODO(BEAM-3304): trigger support | ||
| Trigger *pipepb.Trigger |
There was a problem hiding this comment.
As a rule, we avoid having users have access to, or deal with, the proto directly, including via any of the built up structs. The protos are an implementation detail, not an interface for users to use.
This largely means building up SDK specific mechanisms that are then translated to the proto in graphx/translate.go. Essentially, the graph/** directories shouldn't be depending on the protos.
That said, this is fine during development as Pane plumbing and such is being developed.
|
I changed parts in marshalWindowingStrategy in translate.go to try out triggers. But did undo all before committing. I'll write a separate test doing this. |
sdks/go/pkg/beam/pcollection.go
Outdated
| @@ -49,7 +50,11 @@ func (p PCollection) IsValid() bool { | |||
| } | |||
|
|
|||
| // TODO(herohde) 5/30/2017: add name for PCollections? Java supports it. | |||
|
|
|||
| // TODO(herohde) 5/30/2017: add windowing strategy and documentation. | |||
There was a problem hiding this comment.
Since this becomes part of the public API for PCollection, it should have a real doc comment
// WindowingStrategy returns the windowing strategy for the PCollection
| } | ||
| func makeTrigger(t window.TriggerType) (*pipepb.Trigger, error) { |
There was a problem hiding this comment.
| } | |
| func makeTrigger(t window.TriggerType) (*pipepb.Trigger, error) { | |
| } | |
| func makeTrigger(t window.TriggerType) (*pipepb.Trigger, error) { |
There was a problem hiding this comment.
Probably need to run go fmt over your changes, as for some reason newlines between function declarations seem to be going missing.
|
|
||
| // change below statement to: windowed.WindowingStrategy().SetAlways() | ||
| // to get the decoding error. | ||
| windowed.WindowingStrategy().SetDefault() |
There was a problem hiding this comment.
OK. It pains me to suggest this, but this user experience leaves a little to be desired and we have a short window before the Go SDK exits experimental where we can make it better.
I propose we make a beam.WindowIntoOption interface type, and change beam.WindowInto to take a variadic parameter of those to configure the trigger and other windowing strategy properties. Then we can type assert and apply the options to the windowing strategy without much concern.
This would be similar to how beam.ParDo has an option type https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/option.go#L25
https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/pardo.go#L356
My recommendation would be to make that into a separate change by itself, not defining any of the concrete beam.WindowIntoOption types and handling, and just changing the signature to take in ...options beam.WindowIntoOption as the final parameter (variadic parameters must be the final one in a function signature).
While this change would not be 100% backwards compatible, it would be source compatible for all reasonable uses of beam.WindowInto. The main place it would break is if a user were doing something like:
var myWindowInto func(s Scope, ws *window.Fn, col PCollection) PCollection = beam.WindowInto
which is unlikely and weird to begin with (it wouldn't serve anything beyond ensuring the signature never changes).
Then we can rebase and get back to here and finish handling the implementation.
sdks/go/pkg/beam/core/graph/edge.go
Outdated
| @@ -527,8 +527,8 @@ func NewImpulse(g *Graph, s *Scope, value []byte) *MultiEdge { | |||
| } | |||
|
|
|||
| // NewWindowInto inserts a new WindowInto edge into the graph. | |||
| func NewWindowInto(g *Graph, s *Scope, wfn *window.Fn, in *Node) *MultiEdge { | |||
| n := g.NewNode(in.Type(), &window.WindowingStrategy{Fn: wfn}, in.Bounded()) | |||
| func NewWindowInto(g *Graph, s *Scope, wfn *window.Fn, tr window.TriggerType, in *Node) *MultiEdge { | |||
There was a problem hiding this comment.
NewWindowInto should just change to take in a *window.WindowingStrategy instead of adding more and more parameters to it. There are other modes to handle in the future and by populating the windowing strategy in the Beam package before it gets here, will save time and effort when adding features later.
sdks/go/pkg/beam/windowing.go
Outdated
| @@ -21,21 +21,47 @@ import ( | |||
| "github.com/apache/beam/sdks/go/pkg/beam/internal/errors" | |||
| ) | |||
|
|
|||
| type WindowIntoOption interface { | |||
| private() | |||
There was a problem hiding this comment.
"private" isn't a good name for this blocking function, because beam.Option is already using it. You can test this out yourself by seeing that you can pass WindowIntoOptions as an Option into beam.ParDo.
A better unexported method name would be windowIntoOption() to avoid this conflation.
sdks/go/pkg/beam/windowing.go
Outdated
|
|
||
| func (t WindowTrigger) windowIntoOption() {} | ||
|
|
||
| func (t WindowTrigger) GetName() window.TriggerType { |
There was a problem hiding this comment.
There's no need for a GetName function. As a rule, don't add "Get" to a method that just returns a field.
See https://golang.org/doc/effective_go#Getters
sdks/go/pkg/beam/windowing.go
Outdated
| for _, opt := range opts { | ||
| switch opt.(type) { | ||
| case WindowTrigger: | ||
| edge = graph.NewWindowInto(s.real, s.scope, &window.WindowingStrategy{Fn: ws, Trigger: opt.(WindowTrigger).GetName()}, col.n) |
There was a problem hiding this comment.
Consider accessing the trigger kind directly with .Name without a method instead, and remove the method.
sdks/go/pkg/beam/pcollection.go
Outdated
| // TODO(herohde) 5/30/2017: add windowing strategy and documentation. | ||
|
|
||
| // WindowingStrategy returns the WindowingStrategy of PCollection. | ||
| func (p PCollection) WindowingStrategy() *window.WindowingStrategy { | ||
| return p.n.WindowingStrategy() | ||
| } |
There was a problem hiding this comment.
We can probably leave out this method for now. It's easier to add something like this later than to remove it later.
I'd still remove the TODO. We don't know what herohde's goal was with the TODO, and we've gone in our own design direction.
| // TriggerWindowSums, much like WindowSums described above has an addition of configuring | ||
| // a trigger here. SetDefault works fine. Other triggers such as SetAlways throws | ||
| // pane decoding error. | ||
| func TriggerWindowSums(s beam.Scope, sumPerKey func(beam.Scope, beam.PCollection) beam.PCollection) { |
There was a problem hiding this comment.
Consider re-writing this test to make it easier to validate triggers rather than windows. Certainly re-use what you can, but you are allowed to write new code.
|
|
||
| package window | ||
|
|
||
| type TriggerType string |
There was a problem hiding this comment.
Consider how we'd implement the more complicated triggers that have sub triggers. For example, AfterAll and AfterAny are missing their sub trigger configurations.
| Default TriggerType = "Trigger_Default_" | ||
| Always TriggerType = "Trigger_Always_" | ||
| AfterAny TriggerType = "Trigger_AfterAny_" | ||
| AfterAll TriggerType = "Trigger_AfterAny_" |
There was a problem hiding this comment.
Typo: duplicated as AfterAny instead of AfterAll.
|
PTAL. Added pane helpers and changed all affecting function signatures/receivers. The error of |
There was a problem hiding this comment.
I have many comments, but this is excellent progress. You've found many of the correct places to change without additional guidance.
We'll need maybe one or two more passes after these changes to sort out user side things and further initial testing, but this is great.
|
PTAL. Basic triggers. Will add more tests in windowinto as they are similar to what we currently have. |
|
Resolved the comments and added a test for Repeat Trigger. |
|
PTAL. AfterAll and AfterAny Triggers are throwing Illegal Argument Exception. Working on it. |
|
Run Go PostCommit |
|
Run Go Flink ValidatesRunner |
|
Run Go PostCommit |
|
It's not clear what's causing the failure with flink in this case. It's almost certainly the same TestStream issues. Lets disable the tests on flink entirely for now (like they are for the other runners) and we'll investigate it properly from there. |
|
Run Go Flink ValidatesRunner |
|
Run Go PostCommit |
Trigger support for Go SDK
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
ValidatesRunnercompliance status (on master branch)Examples testing status on various runners
Post-Commit SDK/Transform Integration Tests Status (on master branch)
Pre-Commit Tests Status (on master branch)
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.