jrmccluskey commented on code in PR #38288:
URL: https://github.com/apache/beam/pull/38288#discussion_r4262556551
##########
sdks/go/pkg/beam/core/graph/window/fn.go:
##########
@@ -30,6 +31,7 @@ const (
FixedWindows Kind = "FIX"
SlidingWindows Kind = "SLI"
Sessions Kind = "SES"
+ CustomWindows Kind = "CUS" // User-defined custom WindowFn
Review Comment:
Don't really need the annotated comment here
##########
sdks/go/pkg/beam/core/graph/window/fn.go:
##########
@@ -53,13 +55,51 @@ func NewSessions(gap time.Duration) *Fn {
return &Fn{Kind: Sessions, Gap: gap}
}
+// NewCustom returns a WindowFn backed by a user-defined custom window
+// function. The fn value must be a pointer-to-struct whose concrete type
+// has been registered with [RegisterWindowFn] during init. Custom window
+// functions are non-merging and must return [IntervalWindow] values from
+// their AssignWindows method.
+//
+// NewCustom panics if fn is nil or its type was not registered.
+func NewCustom(fn any) *Fn {
+ if fn == nil {
+ panic("window.NewCustom: fn must not be nil")
+ }
+ t := reflect.TypeOf(fn)
+ st := t
+ if t.Kind() == reflect.Pointer {
+ st = t.Elem()
+ }
+ if _, ok := LookupWindowFn(st); !ok {
+ panic(fmt.Sprintf("window.NewCustom: type %v is not registered;
call window.RegisterWindowFn during init()", t))
+ }
+ return &Fn{Kind: CustomWindows, CustomFn: fn}
+}
Review Comment:
The logic here doesn't necessarily enforce the pointer-to-struct typing
specified in the docstring, consider checking t.Kind() == reflect.Pointer for
input valdiation
##########
CHANGES.md:
##########
@@ -79,6 +79,7 @@
* [Flink Runner] Added opt-in static round-robin split assignment for small
bounded sources via the new `sourceStaticSplitThresholdMb` pipeline option. The
default of 0 keeps the existing lazy pull-based assignment
([#39873](https://github.com/apache/beam/issues/39873)).
* Added automatic caching of bounded, single-pane side-input views for classic
Java Flink DataStream execution
([#39866](https://github.com/apache/beam/issues/39866)).
* (Python) Added `Sample.Any`, the Python equivalent of Java's `Sample.any`,
which returns up to n arbitrary elements from a PCollection
([#18552](https://github.com/apache/beam/issues/18552)).
+* (Go) Added support for custom WindowFns
([#20627](https://github.com/apache/beam/issues/20627)).
Review Comment:
This needs to be moved to the 2.78.0 section during/after a rebase
##########
sdks/go/pkg/beam/core/runtime/exec/window.go:
##########
@@ -82,6 +98,9 @@ func assignWindows(wfn *window.Fn, ts typex.EventTime)
[]typex.Window {
// each other) will be merged.
return []typex.Window{window.IntervalWindow{Start: ts, End:
ts.Add(wfn.Gap)}}
+ case window.CustomWindows:
+ return inv.Invoke(ts, elm, elm2)
Review Comment:
This might silently return nil, we should add an extra check to make sure
this window function returns _something_
##########
sdks/go/pkg/beam/core/graph/window/fn.go:
##########
@@ -53,13 +55,51 @@ func NewSessions(gap time.Duration) *Fn {
return &Fn{Kind: Sessions, Gap: gap}
}
+// NewCustom returns a WindowFn backed by a user-defined custom window
Review Comment:
Should also call out that this feature only works on prism for now since
this feature uses an SDK-specific URN that needs to be implemented per-runner
##########
sdks/go/pkg/beam/core/runtime/graphx/translate.go:
##########
@@ -358,6 +368,11 @@ func getSideWindowMappingUrn(winFn *window.Fn) string {
mappingUrn = URNWindowMappingSliding
case window.Sessions:
panic("session windowing is not supported for side inputs")
+ case window.CustomWindows:
+ if winFn.NeedsElement() {
+ panic("element-aware custom WindowFn is not supported
for side inputs")
+ }
Review Comment:
This check is good, I would suggest adding a similar check in
sdks/go/pkg/beam/pardo.go in TryParDo() where we also check side input windowfn
validity
##########
sdks/go/test/integration/integration.go:
##########
Review Comment:
```suggestion
"TestParDoBundleFinalizer.*",
"TestValidateCustomWindowedSideInputs",
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
