You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "riteshghorse (via GitHub)" <gi...@apache.org> on 2023/04/04 13:32:19 UTC
[GitHub] [beam] riteshghorse opened a new pull request, #26101: [WIP] Timers with new datalayer
riteshghorse opened a new pull request, #26101:
URL: https://github.com/apache/beam/pull/26101
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] 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/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
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)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1176945779
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,227 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string) {
Review Comment:
Thanks for testing it out. I got diverted to other part of the work. I've added support for emitters now. Feel free to test your pipelines
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1522395800
I guess to be fully concrete: The value of the composites won't be realized until we do it to state too (which will mean eventually deprecating the exist litany of methods, but the principle is the same as timers.)
The state change, and the composition helpers don't need to happen in this PR.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1178161580
##########
sdks/go/pkg/beam/core/timers/timers.go:
##########
@@ -0,0 +1,161 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package timers contains structs for setting pipeline timers.
+package timers
+
+import (
+ "reflect"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+)
+
+var (
+ // ProviderType represents the type of timer provider.
+ ProviderType = reflect.TypeOf((*Provider)(nil)).Elem()
+)
+
+// TimeDomainEnum represents different time domains to set timer.
+type TimeDomainEnum int32
+
+const (
+ // TimeDomainUnspecified represents unspecified time domain.
+ TimeDomainUnspecified TimeDomainEnum = 0
+ // TimeDomainEventTime is time from the perspective of the data
+ TimeDomainEventTime TimeDomainEnum = 1
+ // TimeDomainProcessingTime is time from the perspective of the
+ // execution of your pipeline
+ TimeDomainProcessingTime TimeDomainEnum = 2
+)
+
+// TimerMap holds timer information obtained from the pipeline.
+type TimerMap struct {
+ Family string
+ Tag string
+ Clear bool
+ FireTimestamp, HoldTimestamp mtime.Time
+}
+
+type timerConfig struct {
+ Tag string
+ HoldTimestamp mtime.Time
+}
+
+type timerOptions func(*timerConfig)
+
+// WithTag sets the tag for the timer.
+func WithTag(tag string) timerOptions {
+ return func(tm *timerConfig) {
+ tm.Tag = tag
+ }
+}
+
+// WithOutputTimestamp sets the output timestamp for the timer.
+func WithOutputTimestamp(outputTimestamp time.Time) timerOptions {
+ return func(tm *timerConfig) {
+ tm.HoldTimestamp = mtime.FromTime(outputTimestamp)
+ }
+}
+
+// Provider represents a timer provider interface.
+type Provider interface {
+ Set(t TimerMap)
+}
+
+// PipelineTimer interface represents valid timer type.
+type PipelineTimer interface {
+ TimerFamily() string
+ TimerDomain() TimeDomainEnum
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1178321047
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1386,35 +1386,54 @@ func validateState(fn *DoFn, numIn mainInputs) error {
}
func validateOnTimerFn(fn *DoFn) error {
- if _, ok := fn.OnTimerFn(); !ok {
+ if _, ok := fn.methods[onTimerName]; !ok {
err := errors.Errorf("OnTimer function not defined for DoFn: %v", fn.Name())
return errors.SetTopLevelMsgf(err, "OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.", fn.Name())
}
- return nil
-}
+ if _, ok := fn.methods[onTimerName].TimerProvider(); !ok {
+ err := errors.Errorf("OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v", fn.Name(), fn.PipelineTimers())
+ return errors.SetTopLevelMsgf(err, "OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v"+
+ ", Ensure that you are using the TimerProvider to set and clear the timers.", fn.Name(), fn.PipelineTimers())
+ }
-func validateTimer(fn *DoFn) error {
- if fn.Fn == nil {
- return nil
+ _, otNum, otExists := fn.methods[onTimerName].Emits()
+ _, peNum, peExists := fn.methods[processElementName].Emits()
+
+ if otExists == peExists {
+ if otNum != peNum {
+ return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, no. of emitters used in OnTimer: %v, no. of emitters used in ProcessElement: %v", otNum, peNum)
+ }
+ } else {
+ return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, emitters used in OnTimer: %v, emitters used in ProcessElement: %v", otExists, peExists)
}
+ return nil
+}
+
+func validateTimer(fn *DoFn, numIn mainInputs) error {
pt := fn.PipelineTimers()
- if _, ok := fn.Fn.TimerProvider(); ok {
+ if _, ok := fn.methods[processElementName].TimerProvider(); ok {
+ if numIn == MainSingle {
+ err := errors.Errorf("ProcessElement uses a TimerProvider, but is not keyed")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but is not keyed. "+
+ "All stateful DoFns must take a key/value pair as an input.")
+ }
if len(pt) == 0 {
err := errors.New("ProcessElement uses a TimerProvider, but no Timer fields are defined in the DoFn")
return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer fields are defined in the DoFn"+
- ", Ensure that you are including the exported timer field in the DoFn that you're using to set/clear timers.")
+ ", Ensure that your DoFn exports the Timer fields used to set and clear timers.")
}
timerKeys := make(map[string]timers.PipelineTimer)
for _, t := range pt {
- k := t.TimerFamily()
- if timer, ok := timerKeys[k]; ok {
- err := errors.Errorf("Duplicate timer key %v", k)
- return errors.SetTopLevelMsgf(err, "Duplicate timer key %v used by %v and %v. Ensure that keys are unique per DoFn", k, timer, t)
- } else {
- timerKeys[k] = t
+ for timerFamilyID := range t.Timers() {
+ if timer, ok := timerKeys[timerFamilyID]; ok {
+ err := errors.Errorf("Duplicate timer key %v", timerFamilyID)
+ return errors.SetTopLevelMsgf(err, "Duplicate timer family ID %v used by %v and %v. Ensure that timer family IDs are unique per DoFn", timerFamilyID, timer, t)
Review Comment:
the value "timer" here is just a different timer. For this check we may want to store field names in a map, so the error can be sensible. Probably need to change the `PipelineTimers` method to allow for it, by storing both the field name, and the timer for a given family name.
It's good we have an error at least, but
##########
sdks/go/pkg/beam/core/runtime/exec/translate.go:
##########
@@ -588,6 +592,20 @@ func (b *builder) makeLink(from string, id linkID) (Node, error) {
}
}
+ if len(userTimers) > 0 {
+ log.Debugf(context.TODO(), "userTimers %+v", userTimers)
+ timerIDToCoder := make(map[string]*coder.Coder)
+ sID := StreamID{Port: Port{URL: b.desc.GetTimerApiServiceDescriptor().GetUrl()}, PtransformID: id.to}
+ ec, wc, err := b.makeCoderForPCollection(input[0])
+ if err != nil {
+ return nil, err
+ }
+ for key := range userTimers {
+ timerIDToCoder[key] = coder.NewT(ec, wc)
+ }
Review Comment:
2. Yeah, here we don't even use the coderID stored in the spec, but we ultimately pass through the whole PCollection coder instead of extracting out the Key component here.
##########
sdks/go/pkg/beam/core/runtime/exec/timers.go:
##########
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "context"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+)
+
+// UserTimerAdapter provides a timer provider to be used for manipulating timers.
+type UserTimerAdapter interface {
+ NewTimerProvider(ctx context.Context, manager DataManager, inputTimestamp typex.EventTime, windows []typex.Window, element *MainInput) (timerProvider, error)
Review Comment:
5. We can thus also pre-create and cache the Key decoder here, and use adapter for the coder, instead of doing it at the top of the DataSource.
##########
sdks/go/pkg/beam/core/runtime/exec/timers.go:
##########
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "context"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+)
+
+// UserTimerAdapter provides a timer provider to be used for manipulating timers.
+type UserTimerAdapter interface {
+ NewTimerProvider(ctx context.Context, manager DataManager, inputTimestamp typex.EventTime, windows []typex.Window, element *MainInput) (timerProvider, error)
+}
+
+type userTimerAdapter struct {
+ sID StreamID
+ timerIDToCoder map[string]*coder.Coder
+}
+
+// NewUserTimerAdapter returns a user timer adapter for the given StreamID and timer coder.
+func NewUserTimerAdapter(sID StreamID, c *coder.Coder, timerCoders map[string]*coder.Coder) UserTimerAdapter {
+ if !coder.IsW(c) {
+ panic(fmt.Sprintf("expected WV coder for user timer %v: %v", sID, c))
+ }
+
+ return &userTimerAdapter{sID: sID, timerIDToCoder: timerCoders}
+}
+
+// NewTimerProvider creates and returns a timer provider to set/clear timers.
+func (u *userTimerAdapter) NewTimerProvider(ctx context.Context, manager DataManager, inputTs typex.EventTime, w []typex.Window, element *MainInput) (timerProvider, error) {
+ userKey := &FullValue{Elm: element.Key.Elm}
+ tp := timerProvider{
+ ctx: ctx,
+ tm: manager,
+ userKey: userKey,
+ inputTimestamp: inputTs,
+ sID: u.sID,
+ window: w,
+ writersByFamily: make(map[string]io.Writer),
+ codersByFamily: u.timerIDToCoder,
Review Comment:
4. Per this bit of jumping around, since it's always going to be the same coder for all families in the same DoFn, we can pre-make the Encoder here too, so it's not created on every call to Set.
##########
sdks/go/pkg/beam/core/runtime/exec/coder.go:
##########
@@ -145,8 +145,15 @@ func MakeElementEncoder(c *coder.Coder) ElementEncoder {
}
case coder.Timer:
+ tc := coder.SkipW(c).Components[0]
+ if coder.IsKV(tc) {
+ return &timerEncoder{
+ elm: MakeElementEncoder(tc.Components[0]),
+ win: MakeWindowEncoder(c.Window),
+ }
+ }
return &timerEncoder{
- elm: MakeElementEncoder(c.Components[0]),
Review Comment:
3. Yeah, not extracting them early leads to issues at this stage where we only need the key, but we require a KV. Could lead to subtle, weird errors I think.
##########
sdks/go/pkg/beam/pardo.go:
##########
@@ -116,6 +116,22 @@ func TryParDo(s Scope, dofn any, col PCollection, opts ...Option) ([]PCollection
}
}
+ wc := inWfn.Coder()
+ pipelineTimers := fn.PipelineTimers()
+ if len(pipelineTimers) > 0 {
+ c, err := inferCoder(typex.New(reflect.TypeOf(col.Type())))
+ if err != nil {
+ return nil, addParDoCtx(errors.New("error infering coder from col"), s)
+ }
+ edge.TimerCoders = make(map[string]*coder.Coder)
Review Comment:
1. If the coder is always going to be the Key of the input PCollection KV type, do we need to store a different coder for every timer family, since they're identical?
This means we could also avoid infering the coder multiple times, since this could be done in the graphx/translate.go step instead, and we can probably avoid filtering down to the Key type a few times in the exec side. (I should be able to comment on those directly to help track them down...)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hnnsgstfssn commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "hnnsgstfssn (via GitHub)" <gi...@apache.org>.
hnnsgstfssn commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1172506840
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,227 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string) {
Review Comment:
This example only includes the key. If I want to access the value in `OnTimer`, will I have to pass it through the state API, or possibly as a timer tag if it can be made a `string`? I see that [the design](https://docs.google.com/document/d/1rcKa1Z6orDDFr1l8t6NA1eLl6zanQbYAEiAqk39NQUU/) says a value should be mapped, but [this](https://beam.apache.org/documentation/basics/#state-and-timers) says the interaction is through state API. I have not been able to add an input value, getting some variant of
```
panic: reflect: Call using zero Value argument goroutine 1472 [running]:
runtime/debug.Stack()
/usr/lib/go/src/runtime/debug/stack.go:24 +0x65
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.callNoPanic.func1()
/home/rru/go/pkg/mod/github.com/riteshghorse/beam/sdks/v2@v2.0.0-20230420045802-5b012f52afa8/go/pkg/beam/core/runtime/exec/util.go:58 +0xa5
panic({0x1d9c940, 0xc000c39260})
/usr/lib/go/src/runtime/panic.go:884 +0x213
reflect.Value.call({0x2016b60?, 0xc000a4d480?, 0x44f6d2?}, {0x28561b5, 0x4}, {0xc000a89970, 0x7, 0x5ddc11?})
/usr/lib/go/src/reflect/value.go:437 +0x1aee
reflect.Value.Call({0x2016b60?, 0xc000a4d480?, 0x40e427?}, {0xc000a89970?, 0x24e75a0?, 0xc0006e6801?})
/usr/lib/go/src/reflect/value.go:370 +0xbc
github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx.(*reflectFunc).Call(0xc000a86ed0, {0xc000931340?, 0xc0006e6988?, 0xa6660f?})
/home/rru/go/pkg/mod/github.com/riteshghorse/beam/sdks/v2@v2.0.0-20230420045802-5b012f52afa8/go/pkg/beam/core/util/reflectx/call.go:87 +0x59
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*invoker).initCall.func46({0x0, 0x0, 0x0, 0x0, 0x0}, {0x3ea9da0, 0x1, 0x1}, 0xf?)
/home/rru/go/pkg/mod/github.com/riteshghorse/beam/sdks/v2@v2.0.0-20230420045802-5b012f52afa8/go/pkg/beam/core/runtime/exec/fn_arity.go:307 +0x94
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*invoker).invokeWithOpts(0xc000b6a780, {0x2b630a8?, 0xc000a4ccc0}, {0x0, 0x0, 0x0, 0x0, 0x0}, {0x3ea9da0, 0x1, ...}, ...)
```
Similarly I have not been able to add any emitter, getting similar errors as per above.
Another issue I had was that it failed to validate when the `beam.EventTime` was placed after the `timers.Provider`. It says then that it needs to be before the main input. Moving it into the same position as here seems to work.
Finally I was unable to add timers to a GBK DoFn, having it complain about needing to use KV with timers, but I need to investigate further on this one. Maybe it is not supposed to be possible to use timers in such cases?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1179495076
##########
sdks/go/pkg/beam/core/runtime/exec/timers.go:
##########
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "context"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+)
+
+// UserTimerAdapter provides a timer provider to be used for manipulating timers.
+type UserTimerAdapter interface {
+ NewTimerProvider(ctx context.Context, manager DataManager, inputTimestamp typex.EventTime, windows []typex.Window, element *MainInput) (timerProvider, error)
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1178196901
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -263,8 +263,17 @@ func (n *DataSource) Process(ctx context.Context) ([]*Checkpoint, error) {
}
},
func(bcr *byteCountReader, ptransformID, timerFamilyID string) error {
- tmap, err := decodeTimer(cp, wc, bcr)
- log.Infof(ctx, "DEBUGLOG: timer received for: %v and %v - %+v err: %v", ptransformID, timerFamilyID, tmap, err)
+ dc := MakeElementDecoder(coder.SkipW(c).Components[0])
+ tmap, err := decodeTimer(dc, wc, bcr)
+ if err != nil {
+ return errors.WithContext(err, "error decoding timer in datasource")
+ }
+ if fn, ok := n.OnTimerTransforms[ptransformID].Fn.OnTimerFn(); ok {
+ _, err := n.OnTimerTransforms[ptransformID].InvokeTimerFn(ctx, fn, timerFamilyID, tmap)
+ if err != nil {
+ return errors.WithContext(err, "ontimer callback invocation failed")
+ }
+ }
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1179375418
##########
sdks/go/pkg/beam/core/runtime/exec/translate.go:
##########
@@ -588,6 +592,20 @@ func (b *builder) makeLink(from string, id linkID) (Node, error) {
}
}
+ if len(userTimers) > 0 {
+ log.Debugf(context.TODO(), "userTimers %+v", userTimers)
+ timerIDToCoder := make(map[string]*coder.Coder)
+ sID := StreamID{Port: Port{URL: b.desc.GetTimerApiServiceDescriptor().GetUrl()}, PtransformID: id.to}
+ ec, wc, err := b.makeCoderForPCollection(input[0])
+ if err != nil {
+ return nil, err
+ }
+ for key := range userTimers {
+ timerIDToCoder[key] = coder.NewT(ec, wc)
+ }
Review Comment:
Good catch, I agree. We should just send the key coder from here which then simplifies the exec/coder.go for timer encoding/decoding
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1157512442
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1350,6 +1393,42 @@ func validateState(fn *DoFn, numIn mainInputs) error {
return nil
}
+func validateTimer(fn *DoFn) error {
+ if fn.Fn == nil {
+ return nil
+ }
+
+ pt := fn.PipelineTimers()
+
+ if _, ok := fn.Fn.TimerProvider(); ok {
+ if len(pt) == 0 {
+ err := errors.Errorf("ProcessElement uses a TimerProvider, but no timer struct-tags are attached to the DoFn")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer struct-tags are attached to the DoFn"+
+ ", Ensure that you are including the timer structs you're using to set/clear global state as uppercase member variables")
+ }
+ timerKeys := make(map[string]PipelineTimer)
+ for _, t := range pt {
+ k := t.TimerFamily()
+ if timer, ok := timerKeys[k]; ok {
+ err := errors.Errorf("Duplicate timer key %v", k)
+ return errors.SetTopLevelMsgf(err, "Duplicate timer key %v used by %v and %v. Ensure that keys are unique per DoFn", k, timer, t)
+ } else {
+ timerKeys[k] = t
+ }
+ }
+ } else {
+ if len(pt) > 0 {
+ err := errors.Errorf("ProcessElement doesn't use a TimerProvider, but Timer Struct is attached to the DoFn: %v", pt)
+ return errors.SetTopLevelMsgf(err, "ProcessElement doesn't use a TimerProvider, but Timer Struct is attached to the DoFn: %v"+
Review Comment:
"but the DoFn has no Timer fields"...
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1350,6 +1393,42 @@ func validateState(fn *DoFn, numIn mainInputs) error {
return nil
}
+func validateTimer(fn *DoFn) error {
+ if fn.Fn == nil {
+ return nil
+ }
+
+ pt := fn.PipelineTimers()
+
+ if _, ok := fn.Fn.TimerProvider(); ok {
+ if len(pt) == 0 {
+ err := errors.Errorf("ProcessElement uses a TimerProvider, but no timer struct-tags are attached to the DoFn")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer struct-tags are attached to the DoFn"+
Review Comment:
Technically we're just looking for exported Timer fields, not struct tags. struct tags are the fixed strings after the timer fields.
I'm not against using them instead of needing to name the fields and so forth, since then users don't need to initialise the fields themselves to get the names but that approach would prevent the "Composite" idea I have, since we can't... extrospect? the containing struct from the field. Pros and and Cons...
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -265,6 +265,30 @@ func (n *DataSource) Process(ctx context.Context) ([]*Checkpoint, error) {
func(bcr *byteCountReader, ptransformID, timerFamilyID string) error {
tmap, err := decodeTimer(cp, wc, bcr)
log.Infof(ctx, "DEBUGLOG: timer received for: %v and %v - %+v err: %v", ptransformID, timerFamilyID, tmap, err)
+ log.Infof(ctx, "OnTimerTransforms = %+v", n.OnTimerTransforms[ptransformID].Fn)
+ if fn, ok := n.OnTimerTransforms[ptransformID].Fn.OnTimerFn(); ok {
+ log.Infof(ctx, "found ontimer method, invoking callback")
+ _, err := n.OnTimerTransforms[ptransformID].InvokeTimerFn(ctx, fn, timerFamilyID, tmap)
+ if err != nil {
+ return errors.WithContext(err, "ontimer callback invocation failed")
+ }
+ }
+ // Check if there's a continuation and return residuals
+ // Needs to be done immediately after processing to not lose the element.
+ // if c := n.getProcessContinuation(); c != nil {
+ // cp, err := n.checkpointThis(ctx, c)
+ // if err != nil {
+ // // Errors during checkpointing should fail a bundle.
+ // return err
+ // }
+ // if cp != nil {
+ // checkpoints = append(checkpoints, cp)
+ // }
+ // }
+ // We've finished processing an element, check if we have finished a split.
+ // if n.incrementIndexAndCheckSplit() {
+ // return splitSuccess
+ // }
Review Comment:
I'm 99% sure that OnTimer shouldn't deal with Process Continuations.
I think we'd just want to be able to set another timer?
##########
sdks/go/pkg/beam/core/runtime/exec/timers.go:
##########
@@ -0,0 +1,132 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "context"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+type UserTimerAdapter interface {
+ NewTimerProvider(ctx context.Context, manager DataManager, inputTimestamp typex.EventTime, windows []typex.Window, element *MainInput) (timerProvider, error)
+}
+
+type userTimerAdapter struct {
+ SID StreamID
+ wc WindowEncoder
+ kc ElementEncoder
+ Dc ElementDecoder
+ TimerIDToCoder map[string]*coder.Coder
+ C *coder.Coder
+}
+
+func NewUserTimerAdapter(sID StreamID, c *coder.Coder, timerCoders map[string]*coder.Coder) UserTimerAdapter {
+ if !coder.IsW(c) {
+ panic(fmt.Sprintf("expected WV coder for user timer %v: %v", sID, c))
+ }
+
+ wc := MakeWindowEncoder(c.Window)
+ var kc ElementEncoder
+ var dc ElementDecoder
+ if coder.IsKV(coder.SkipW(c)) {
Review Comment:
Note: if it isn't a KV, this should fail at pipeline construction time. Timers (and State) are always keyed, so at execution time, we should be able to take that as a guarantee and not have to check the coder at every layer WRT timers.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1170796423
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -265,6 +265,30 @@ func (n *DataSource) Process(ctx context.Context) ([]*Checkpoint, error) {
func(bcr *byteCountReader, ptransformID, timerFamilyID string) error {
tmap, err := decodeTimer(cp, wc, bcr)
log.Infof(ctx, "DEBUGLOG: timer received for: %v and %v - %+v err: %v", ptransformID, timerFamilyID, tmap, err)
+ log.Infof(ctx, "OnTimerTransforms = %+v", n.OnTimerTransforms[ptransformID].Fn)
+ if fn, ok := n.OnTimerTransforms[ptransformID].Fn.OnTimerFn(); ok {
+ log.Infof(ctx, "found ontimer method, invoking callback")
+ _, err := n.OnTimerTransforms[ptransformID].InvokeTimerFn(ctx, fn, timerFamilyID, tmap)
+ if err != nil {
+ return errors.WithContext(err, "ontimer callback invocation failed")
+ }
+ }
+ // Check if there's a continuation and return residuals
+ // Needs to be done immediately after processing to not lose the element.
+ // if c := n.getProcessContinuation(); c != nil {
+ // cp, err := n.checkpointThis(ctx, c)
+ // if err != nil {
+ // // Errors during checkpointing should fail a bundle.
+ // return err
+ // }
+ // if cp != nil {
+ // checkpoints = append(checkpoints, cp)
+ // }
+ // }
+ // We've finished processing an element, check if we have finished a split.
+ // if n.incrementIndexAndCheckSplit() {
+ // return splitSuccess
+ // }
Review Comment:
Done, thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1174021972
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,227 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string) {
Review Comment:
Thanks for your interest!
The design doc is a little out of date, and was authored with a less than complete understanding of timers. The Beam Model doc is correct and any values would need to be carried over via the state API.
The FnAPI simply doesn't have a provision for the additional data.
Timers and State are implicitly per key, but timers can have an additional "tag" associated with them for an additional uniqueness factor.
--------
We definitely do need to have emitters working (which would always match the ProcessElement emitters exactly), and it's probably simply just not implemented in this WIP PR. Adding new "lifecycle" methods isn't simple.
I believe that GBKs *should* be able to have a timer callback and associated state it's largely just a validation thing on the SDK that needs updating.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1522391421
Stopping reviewer notifications for this pull request: requested by reviewer
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1170796960
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1350,6 +1393,42 @@ func validateState(fn *DoFn, numIn mainInputs) error {
return nil
}
+func validateTimer(fn *DoFn) error {
+ if fn.Fn == nil {
+ return nil
+ }
+
+ pt := fn.PipelineTimers()
+
+ if _, ok := fn.Fn.TimerProvider(); ok {
+ if len(pt) == 0 {
+ err := errors.Errorf("ProcessElement uses a TimerProvider, but no timer struct-tags are attached to the DoFn")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer struct-tags are attached to the DoFn"+
Review Comment:
Yeah.. it should be timer fields.. wrote this error message a long ago when we were thinking of struct tags
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1522316046
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @jrmccluskey for label go.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1527547836
Thanks for the reviews and the very important data layer rewrite!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1157507572
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -304,6 +308,40 @@ func (f *DoFn) PipelineState() []state.PipelineState {
return s
}
+type PipelineTimer interface {
+ TimerFamily() string
+ TimerDomain() timers.TimeDomainEnum
+}
Review Comment:
What do you think about switching this to just a "Timers() map[string]timer.Domain" method, to enable the composites I proposed? https://github.com/apache/beam/issues/25894
We wouldn't need to make the State change in this PR (way way out of scope), but now is ideal for enabling this for timers.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hnnsgstfssn commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "hnnsgstfssn (via GitHub)" <gi...@apache.org>.
hnnsgstfssn commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1172506840
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,227 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string) {
Review Comment:
This example only includes the key. If I want to access the value in `OnTimer`, will I have to pass it through the state API? I see that [the design](https://docs.google.com/document/d/1rcKa1Z6orDDFr1l8t6NA1eLl6zanQbYAEiAqk39NQUU/) says a value should be mapped, but [this](https://beam.apache.org/documentation/basics/#state-and-timers) says the interaction is through state API. I have not been able to add an input value, getting some variant of
```
panic: reflect: Call using zero Value argument goroutine 1472 [running]:
runtime/debug.Stack()
/usr/lib/go/src/runtime/debug/stack.go:24 +0x65
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.callNoPanic.func1()
/home/rru/go/pkg/mod/github.com/riteshghorse/beam/sdks/v2@v2.0.0-20230420045802-5b012f52afa8/go/pkg/beam/core/runtime/exec/util.go:58 +0xa5
panic({0x1d9c940, 0xc000c39260})
/usr/lib/go/src/runtime/panic.go:884 +0x213
reflect.Value.call({0x2016b60?, 0xc000a4d480?, 0x44f6d2?}, {0x28561b5, 0x4}, {0xc000a89970, 0x7, 0x5ddc11?})
/usr/lib/go/src/reflect/value.go:437 +0x1aee
reflect.Value.Call({0x2016b60?, 0xc000a4d480?, 0x40e427?}, {0xc000a89970?, 0x24e75a0?, 0xc0006e6801?})
/usr/lib/go/src/reflect/value.go:370 +0xbc
github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx.(*reflectFunc).Call(0xc000a86ed0, {0xc000931340?, 0xc0006e6988?, 0xa6660f?})
/home/rru/go/pkg/mod/github.com/riteshghorse/beam/sdks/v2@v2.0.0-20230420045802-5b012f52afa8/go/pkg/beam/core/util/reflectx/call.go:87 +0x59
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*invoker).initCall.func46({0x0, 0x0, 0x0, 0x0, 0x0}, {0x3ea9da0, 0x1, 0x1}, 0xf?)
/home/rru/go/pkg/mod/github.com/riteshghorse/beam/sdks/v2@v2.0.0-20230420045802-5b012f52afa8/go/pkg/beam/core/runtime/exec/fn_arity.go:307 +0x94
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*invoker).invokeWithOpts(0xc000b6a780, {0x2b630a8?, 0xc000a4ccc0}, {0x0, 0x0, 0x0, 0x0, 0x0}, {0x3ea9da0, 0x1, ...}, ...)
```
Similarly I have not been able to add any emitter, getting similar errors as per above.
Another issue I had was that it failed to validate when the `beam.EventTime` was placed after the `timers.Provider`. It says then that it needs to be before the main input. Moving it into the same position as here seems to work.
Finally I was unable to add timers to a GBK DoFn, having it complain about needing to use KV with timers, but I need to investigate further on this one. Maybe it is not supposed to be possible to use timers in such cases?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on pull request #26101: [WIP] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1522313229
@lostluck could you also take a look now?
Regarding the composites, are you proposing a single interface method that would return map[string]TimeDomain? And this method would return just a single KV pair for the built-in timer types (event-time, processing-time) but would be customizable by users in case they want to batch them.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1523818449
Thanks for the detailed review @lostluck . I've addressed the comments, PTAL
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1179655659
##########
sdks/go/pkg/beam/pardo.go:
##########
@@ -116,6 +116,22 @@ func TryParDo(s Scope, dofn any, col PCollection, opts ...Option) ([]PCollection
}
}
+ wc := inWfn.Coder()
+ pipelineTimers := fn.PipelineTimers()
+ if len(pipelineTimers) > 0 {
+ c, err := inferCoder(typex.New(reflect.TypeOf(col.Type())))
+ if err != nil {
+ return nil, addParDoCtx(errors.New("error infering coder from col"), s)
+ }
+ edge.TimerCoders = make(map[string]*coder.Coder)
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1526470775
Done the coder related refactoring, PTAL
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse merged pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse merged PR #26101:
URL: https://github.com/apache/beam/pull/26101
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1177040045
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
Review Comment:
Just use `time.Now()` directly here, since it's getting converted to a `time.Time` anyway.
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
+ emit(timerKey, timerTag)
+ }
+ }
+}
+
+func (s *Stateful) ProcessElement(ctx context.Context, ts beam.EventTime, sp state.Provider, tp timers.Provider, key, word string, emit func(string, string)) error {
+ s.ElementBag.Add(sp, word)
+ s.MinTime.Add(sp, int64(ts))
+
+ toFire, ok, err := s.TimerTime.Read(sp)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ toFire = int64(mtime.Now().Add(1 * time.Minute))
+ }
+ minTime, _, err := s.MinTime.Read(sp)
+ if err != nil {
+ return err
+ }
+
+ s.OutputState.Set(tp, mtime.Time(toFire).ToTime(), timers.WithOutputTimestamp(mtime.Time(minTime).ToTime()), timers.WithTag(word))
+ s.TimerTime.Write(sp, toFire)
+
+ return nil
+}
+
+type eventtimeSDFStream struct {
Review Comment:
```suggestion
// eventtimeSDFStream is a Watermark Estimating, Unbounded Splittable DoFn that emits a single element, then sleeps until the next emit by the runner.
// After emitting the configured number of elements, it stops processing, allowing the pipeline to terminate.
type eventtimeSDFStream struct {
```
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
Review Comment:
Once the changes have been done to the pipeline, we can update this Description to match it (eg. it's using periodic sequence, and it will self terminate, so we don't need to tell people to cancel it, etc).
##########
sdks/go/pkg/beam/core/runtime/exec/pardo.go:
##########
@@ -345,6 +355,41 @@ func (n *ParDo) invokeDataFn(ctx context.Context, pn typex.PaneInfo, ws []typex.
return val, nil
}
+func (n *ParDo) InvokeTimerFn(ctx context.Context, fn *funcx.Fn, timerFamilyID string, tmap TimerRecv) (*FullValue, error) {
+ // Defer side input clean-up in case of panic
+ var err error
+ defer func() {
+ if postErr := n.postInvoke(); postErr != nil {
+ err = postErr
+ }
+ }()
+ if err := n.preInvoke(ctx, tmap.Windows, tmap.HoldTimestamp); err != nil {
+ return nil, err
+ }
+
+ var extra []any
+ extra = append(extra, timerFamilyID)
+
+ if tmap.Tag != "" {
+ extra = append(extra, tmap.Tag)
+ }
+ extra = append(extra, n.cache.extra...)
+ val, err := InvokeWithOpts(ctx, fn, tmap.Pane, tmap.Windows, tmap.HoldTimestamp, InvokeOpts{
+ opt: &MainInput{Key: *tmap.Key},
+ bf: n.bf,
+ we: n.we,
+ sa: n.UState,
+ sr: n.reader,
+ ta: n.Timer,
+ tm: n.timerManager,
+ extra: extra,
+ })
Review Comment:
Please create and cache the onTimer invoker, like we do for processElement.
It's likely that with the move to more streaming and the single element bundles, we'll want to *(not now but later)* optimize the StartBundle and FinishBundle calls too. But I'll need to get a performance measure before doing that.
Besides, it's looking like there's a faster & zero alloc `reflect.Caller` thing that may land with Go 1.21. If that performs well enough, that work (and all the generic generated stuff), might become moot. *fingers crossed*
##########
sdks/go/pkg/beam/core/timers/timers.go:
##########
@@ -0,0 +1,161 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package timers contains structs for setting pipeline timers.
+package timers
+
+import (
+ "reflect"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+)
+
+var (
+ // ProviderType represents the type of timer provider.
+ ProviderType = reflect.TypeOf((*Provider)(nil)).Elem()
+)
+
+// TimeDomainEnum represents different time domains to set timer.
+type TimeDomainEnum int32
+
+const (
+ // TimeDomainUnspecified represents unspecified time domain.
+ TimeDomainUnspecified TimeDomainEnum = 0
+ // TimeDomainEventTime is time from the perspective of the data
+ TimeDomainEventTime TimeDomainEnum = 1
+ // TimeDomainProcessingTime is time from the perspective of the
+ // execution of your pipeline
+ TimeDomainProcessingTime TimeDomainEnum = 2
+)
+
+// TimerMap holds timer information obtained from the pipeline.
+type TimerMap struct {
+ Family string
+ Tag string
+ Clear bool
+ FireTimestamp, HoldTimestamp mtime.Time
+}
+
+type timerConfig struct {
+ Tag string
+ HoldTimestamp mtime.Time
+}
+
+type timerOptions func(*timerConfig)
+
+// WithTag sets the tag for the timer.
+func WithTag(tag string) timerOptions {
+ return func(tm *timerConfig) {
+ tm.Tag = tag
+ }
+}
+
+// WithOutputTimestamp sets the output timestamp for the timer.
+func WithOutputTimestamp(outputTimestamp time.Time) timerOptions {
+ return func(tm *timerConfig) {
+ tm.HoldTimestamp = mtime.FromTime(outputTimestamp)
+ }
+}
+
+// Provider represents a timer provider interface.
+type Provider interface {
+ Set(t TimerMap)
+}
+
+// PipelineTimer interface represents valid timer type.
+type PipelineTimer interface {
+ TimerFamily() string
+ TimerDomain() TimeDomainEnum
Review Comment:
To enable composites, basically these two methods are replaced with `BeamTimers() map[string]TimerDomain` and similarly for the implementations returning like `return map[string]TimerDomainEnum`
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
Review Comment:
Remove, these are unused in this example.
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
+ emit(timerKey, timerTag)
+ }
+ }
+}
+
+func (s *Stateful) ProcessElement(ctx context.Context, ts beam.EventTime, sp state.Provider, tp timers.Provider, key, word string, emit func(string, string)) error {
+ s.ElementBag.Add(sp, word)
+ s.MinTime.Add(sp, int64(ts))
+
+ toFire, ok, err := s.TimerTime.Read(sp)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ toFire = int64(mtime.Now().Add(1 * time.Minute))
+ }
+ minTime, _, err := s.MinTime.Read(sp)
+ if err != nil {
+ return err
+ }
+
+ s.OutputState.Set(tp, mtime.Time(toFire).ToTime(), timers.WithOutputTimestamp(mtime.Time(minTime).ToTime()), timers.WithTag(word))
+ s.TimerTime.Write(sp, toFire)
+
+ return nil
+}
+
+type eventtimeSDFStream struct {
+ RestSize, Mod, Fixed int64
+ Sleep time.Duration
+}
+
+func (fn *eventtimeSDFStream) Setup() error {
+ return nil
+}
+
+func (fn *eventtimeSDFStream) CreateInitialRestriction(v beam.T) offsetrange.Restriction {
+ return offsetrange.Restriction{Start: 0, End: fn.RestSize}
+}
+
+func (fn *eventtimeSDFStream) SplitRestriction(v beam.T, r offsetrange.Restriction) []offsetrange.Restriction {
+ // No split
+ return []offsetrange.Restriction{r}
+}
+
+func (fn *eventtimeSDFStream) RestrictionSize(v beam.T, r offsetrange.Restriction) float64 {
+ return r.Size()
+}
+
+func (fn *eventtimeSDFStream) CreateTracker(r offsetrange.Restriction) *sdf.LockRTracker {
+ return sdf.NewLockRTracker(offsetrange.NewTracker(r))
+}
+
+func (fn *eventtimeSDFStream) ProcessElement(ctx context.Context, _ *CWE, rt *sdf.LockRTracker, v beam.T, emit func(beam.EventTime, int64)) sdf.ProcessContinuation {
+ r := rt.GetRestriction().(offsetrange.Restriction)
+ i := r.Start
+ if r.Size() < 1 {
+ log.Debugf(ctx, "size 0 restriction, stoping to process sentinel %v", slog.Any("value", v))
+ return sdf.StopProcessing()
+ }
+ slog.Debug("emitting element to restriction", slog.Any("value", v), slog.Group("restriction",
+ slog.Any("value", v),
+ slog.Float64("size", r.Size()),
+ slog.Int64("pos", i),
+ ))
+ if rt.TryClaim(i) {
+ v := (i % fn.Mod) + fn.Fixed
+ emit(mtime.Now(), v)
+ }
+ return sdf.ResumeProcessingIn(fn.Sleep)
+}
+
+func (fn *eventtimeSDFStream) InitialWatermarkEstimatorState(_ beam.EventTime, _ offsetrange.Restriction, _ beam.T) int64 {
+ return int64(mtime.MinTimestamp)
+}
+
+func (fn *eventtimeSDFStream) CreateWatermarkEstimator(initialState int64) *CWE {
+ return &CWE{Watermark: initialState}
+}
+
+func (fn *eventtimeSDFStream) WatermarkEstimatorState(e *CWE) int64 {
+ return e.Watermark
+}
+
+type CWE struct {
+ Watermark int64 // uses int64, since the SDK prevent mtime.Time from serialization.
+}
+
+func (e *CWE) CurrentWatermark() time.Time {
+ return mtime.Time(e.Watermark).ToTime()
+}
+
+func (e *CWE) ObserveTimestamp(ts time.Time) {
+ // We add 10 milliseconds to allow window boundaries to
+ // progress after emitting
+ e.Watermark = int64(mtime.FromTime(ts.Add(-90 * time.Millisecond)))
+}
+
+func init() {
+ register.DoFn7x1[context.Context, beam.EventTime, state.Provider, timers.Provider, string, string, func(string, string), error](&Stateful{})
+ register.Emitter2[string, string]()
+ register.DoFn5x1[context.Context, *CWE, *sdf.LockRTracker, beam.T, func(beam.EventTime, int64), sdf.ProcessContinuation]((*eventtimeSDFStream)(nil))
+ register.Emitter2[beam.EventTime, int64]()
+}
+
+func main() {
+ flag.Parse()
+ beam.Init()
+
+ ctx := context.Background()
+
+ log.Infof(ctx, "Publishing %v messages to: %v", len(data), *input)
Review Comment:
Remove, we don't publish to pubsub in this example.
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
+ emit(timerKey, timerTag)
+ }
+ }
+}
+
+func (s *Stateful) ProcessElement(ctx context.Context, ts beam.EventTime, sp state.Provider, tp timers.Provider, key, word string, emit func(string, string)) error {
+ s.ElementBag.Add(sp, word)
+ s.MinTime.Add(sp, int64(ts))
+
+ toFire, ok, err := s.TimerTime.Read(sp)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ toFire = int64(mtime.Now().Add(1 * time.Minute))
+ }
+ minTime, _, err := s.MinTime.Read(sp)
+ if err != nil {
+ return err
+ }
+
+ s.OutputState.Set(tp, mtime.Time(toFire).ToTime(), timers.WithOutputTimestamp(mtime.Time(minTime).ToTime()), timers.WithTag(word))
+ s.TimerTime.Write(sp, toFire)
+
+ return nil
+}
+
+type eventtimeSDFStream struct {
+ RestSize, Mod, Fixed int64
+ Sleep time.Duration
+}
+
+func (fn *eventtimeSDFStream) Setup() error {
+ return nil
+}
+
+func (fn *eventtimeSDFStream) CreateInitialRestriction(v beam.T) offsetrange.Restriction {
+ return offsetrange.Restriction{Start: 0, End: fn.RestSize}
+}
+
+func (fn *eventtimeSDFStream) SplitRestriction(v beam.T, r offsetrange.Restriction) []offsetrange.Restriction {
+ // No split
+ return []offsetrange.Restriction{r}
+}
+
+func (fn *eventtimeSDFStream) RestrictionSize(v beam.T, r offsetrange.Restriction) float64 {
+ return r.Size()
+}
+
+func (fn *eventtimeSDFStream) CreateTracker(r offsetrange.Restriction) *sdf.LockRTracker {
+ return sdf.NewLockRTracker(offsetrange.NewTracker(r))
+}
+
+func (fn *eventtimeSDFStream) ProcessElement(ctx context.Context, _ *CWE, rt *sdf.LockRTracker, v beam.T, emit func(beam.EventTime, int64)) sdf.ProcessContinuation {
+ r := rt.GetRestriction().(offsetrange.Restriction)
+ i := r.Start
+ if r.Size() < 1 {
+ log.Debugf(ctx, "size 0 restriction, stoping to process sentinel %v", slog.Any("value", v))
Review Comment:
We can change this to an log.Infof, so it shows up in logs easily. It's a toy example, so it's nice to have indications of the benefit.
##########
sdks/go/pkg/beam/core/funcx/fn.go:
##########
Review Comment:
Please add unit tests for some of the new paths in the state machine & the TimerProvider method to fn_test.go
https://github.com/apache/beam/blob/601dd58a5763e75c06c71cf98ae7c2d4c9591f8d/sdks/go/pkg/beam/core/funcx/fn_test.go#L51
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
Review Comment:
Same here, Unit tests: https://github.com/apache/beam/blob/601dd58a5763e75c06c71cf98ae7c2d4c9591f8d/sdks/go/pkg/beam/core/graph/fn_test.go
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1350,6 +1385,57 @@ func validateState(fn *DoFn, numIn mainInputs) error {
return nil
}
+func validateOnTimerFn(fn *DoFn) error {
+ if _, ok := fn.OnTimerFn(); !ok {
+ err := errors.Errorf("OnTimer function not defined for DoFn: %v", fn.Name())
+ return errors.SetTopLevelMsgf(err, "OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.", fn.Name())
+ }
+
+ return nil
+}
+
+func validateTimer(fn *DoFn) error {
+ if fn.Fn == nil {
+ return nil
+ }
+
+ pt := fn.PipelineTimers()
+
+ if _, ok := fn.Fn.TimerProvider(); ok {
+ if len(pt) == 0 {
+ err := errors.New("ProcessElement uses a TimerProvider, but no Timer fields are defined in the DoFn")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer fields are defined in the DoFn"+
+ ", Ensure that you are including the exported timer field in the DoFn that you're using to set/clear timers.")
Review Comment:
```suggestion
", Ensure that your DoFn exports the Timer fields used to set and clear timers.")
```
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1350,6 +1385,57 @@ func validateState(fn *DoFn, numIn mainInputs) error {
return nil
}
+func validateOnTimerFn(fn *DoFn) error {
+ if _, ok := fn.OnTimerFn(); !ok {
+ err := errors.Errorf("OnTimer function not defined for DoFn: %v", fn.Name())
+ return errors.SetTopLevelMsgf(err, "OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.", fn.Name())
+ }
+
Review Comment:
This should validate that any emitters match ProcessElement exactly.
It doesn't seem like OnTimer should support side inputs. Reuven has mentioned the semantics would be unclear (eg. which timers etc).
##########
sdks/go/pkg/beam/core/runtime/exec/coder.go:
##########
@@ -1282,20 +1299,19 @@ func encodeTimer(elm ElementEncoder, win WindowEncoder, tm typex.TimerMap, w io.
return errors.WithContext(err, "error encoding paneinfo")
}
}
-
w.Write(b.Bytes())
+
return nil
}
// decodeTimer decodes timer byte encoded with standard timer coder spec.
-func decodeTimer(dec ElementDecoder, win WindowDecoder, r io.Reader) (typex.TimerMap, error) {
- tm := typex.TimerMap{}
-
- fv, err := dec.Decode(r)
+func decodeTimer(dec ElementDecoder, win WindowDecoder, r io.Reader) (TimerRecv, error) {
+ tm := TimerRecv{}
+ key, err := dec.Decode(r)
if err != nil {
- return tm, errors.WithContext(err, "error decoding timer key")
Review Comment:
I think we can keep the "timer key" here, since it's refering to the key that this timer is associated with. (distinct from family)
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -263,8 +263,17 @@ func (n *DataSource) Process(ctx context.Context) ([]*Checkpoint, error) {
}
},
func(bcr *byteCountReader, ptransformID, timerFamilyID string) error {
- tmap, err := decodeTimer(cp, wc, bcr)
- log.Infof(ctx, "DEBUGLOG: timer received for: %v and %v - %+v err: %v", ptransformID, timerFamilyID, tmap, err)
+ dc := MakeElementDecoder(coder.SkipW(c).Components[0])
Review Comment:
Move this out of the per element loop please. It's better that it's accidentally called all the time because of single element bundles, than called on every individual timer in a single batch.
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
+ emit(timerKey, timerTag)
+ }
+ }
+}
+
+func (s *Stateful) ProcessElement(ctx context.Context, ts beam.EventTime, sp state.Provider, tp timers.Provider, key, word string, emit func(string, string)) error {
+ s.ElementBag.Add(sp, word)
+ s.MinTime.Add(sp, int64(ts))
+
+ toFire, ok, err := s.TimerTime.Read(sp)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ toFire = int64(mtime.Now().Add(1 * time.Minute))
+ }
+ minTime, _, err := s.MinTime.Read(sp)
+ if err != nil {
+ return err
+ }
+
+ s.OutputState.Set(tp, mtime.Time(toFire).ToTime(), timers.WithOutputTimestamp(mtime.Time(minTime).ToTime()), timers.WithTag(word))
+ s.TimerTime.Write(sp, toFire)
+
+ return nil
+}
+
+type eventtimeSDFStream struct {
+ RestSize, Mod, Fixed int64
+ Sleep time.Duration
+}
+
+func (fn *eventtimeSDFStream) Setup() error {
+ return nil
+}
+
+func (fn *eventtimeSDFStream) CreateInitialRestriction(v beam.T) offsetrange.Restriction {
+ return offsetrange.Restriction{Start: 0, End: fn.RestSize}
+}
+
+func (fn *eventtimeSDFStream) SplitRestriction(v beam.T, r offsetrange.Restriction) []offsetrange.Restriction {
+ // No split
+ return []offsetrange.Restriction{r}
+}
+
+func (fn *eventtimeSDFStream) RestrictionSize(v beam.T, r offsetrange.Restriction) float64 {
+ return r.Size()
+}
+
+func (fn *eventtimeSDFStream) CreateTracker(r offsetrange.Restriction) *sdf.LockRTracker {
+ return sdf.NewLockRTracker(offsetrange.NewTracker(r))
+}
+
+func (fn *eventtimeSDFStream) ProcessElement(ctx context.Context, _ *CWE, rt *sdf.LockRTracker, v beam.T, emit func(beam.EventTime, int64)) sdf.ProcessContinuation {
+ r := rt.GetRestriction().(offsetrange.Restriction)
+ i := r.Start
+ if r.Size() < 1 {
+ log.Debugf(ctx, "size 0 restriction, stoping to process sentinel %v", slog.Any("value", v))
+ return sdf.StopProcessing()
+ }
+ slog.Debug("emitting element to restriction", slog.Any("value", v), slog.Group("restriction",
+ slog.Any("value", v),
+ slog.Float64("size", r.Size()),
+ slog.Int64("pos", i),
+ ))
Review Comment:
We need to remove the references to `slog` in this example for the time being. It's not yet part of the standard library, and doesn't currently add value over the beam log package.
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1350,6 +1385,57 @@ func validateState(fn *DoFn, numIn mainInputs) error {
return nil
}
+func validateOnTimerFn(fn *DoFn) error {
+ if _, ok := fn.OnTimerFn(); !ok {
+ err := errors.Errorf("OnTimer function not defined for DoFn: %v", fn.Name())
+ return errors.SetTopLevelMsgf(err, "OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.", fn.Name())
+ }
+
+ return nil
+}
+
+func validateTimer(fn *DoFn) error {
+ if fn.Fn == nil {
+ return nil
+ }
+
+ pt := fn.PipelineTimers()
+
+ if _, ok := fn.Fn.TimerProvider(); ok {
+ if len(pt) == 0 {
+ err := errors.New("ProcessElement uses a TimerProvider, but no Timer fields are defined in the DoFn")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer fields are defined in the DoFn"+
+ ", Ensure that you are including the exported timer field in the DoFn that you're using to set/clear timers.")
+ }
+ timerKeys := make(map[string]timers.PipelineTimer)
+ for _, t := range pt {
+ k := t.TimerFamily()
+ if timer, ok := timerKeys[k]; ok {
+ err := errors.Errorf("Duplicate timer key %v", k)
+ return errors.SetTopLevelMsgf(err, "Duplicate timer key %v used by %v and %v. Ensure that keys are unique per DoFn", k, timer, t)
Review Comment:
We should have the error match the Programming Guide term, "timer family ID" not "timer key".
```suggestion
err := errors.Errorf("Duplicate timer family ID %v", k)
return errors.SetTopLevelMsgf(err, "Duplicate timer family ID %v used by %v and %v. Ensure that timer family IDs are unique per DoFn", k, timer, t)
```
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -263,8 +263,17 @@ func (n *DataSource) Process(ctx context.Context) ([]*Checkpoint, error) {
}
},
func(bcr *byteCountReader, ptransformID, timerFamilyID string) error {
- tmap, err := decodeTimer(cp, wc, bcr)
- log.Infof(ctx, "DEBUGLOG: timer received for: %v and %v - %+v err: %v", ptransformID, timerFamilyID, tmap, err)
+ dc := MakeElementDecoder(coder.SkipW(c).Components[0])
+ tmap, err := decodeTimer(dc, wc, bcr)
+ if err != nil {
+ return errors.WithContext(err, "error decoding timer in datasource")
+ }
+ if fn, ok := n.OnTimerTransforms[ptransformID].Fn.OnTimerFn(); ok {
+ _, err := n.OnTimerTransforms[ptransformID].InvokeTimerFn(ctx, fn, timerFamilyID, tmap)
+ if err != nil {
+ return errors.WithContext(err, "ontimer callback invocation failed")
+ }
+ }
Review Comment:
It's worth having a WARN log here saying that `"expected transform <PTransformID> to have an OnTimer method to accept "<timerFamilyID>" callback, but it did not. Please file an issue with Apache Beam with a reproducing DoFn: https://github.com/apache/beam/issues"`
##########
sdks/go/pkg/beam/core/timers/timers.go:
##########
@@ -0,0 +1,161 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package timers contains structs for setting pipeline timers.
+package timers
+
+import (
+ "reflect"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+)
+
+var (
+ // ProviderType represents the type of timer provider.
+ ProviderType = reflect.TypeOf((*Provider)(nil)).Elem()
+)
+
+// TimeDomainEnum represents different time domains to set timer.
+type TimeDomainEnum int32
Review Comment:
I'd just call this "TimeDomain" instead of saying it's an enum.
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
+ emit(timerKey, timerTag)
+ }
+ }
+}
+
+func (s *Stateful) ProcessElement(ctx context.Context, ts beam.EventTime, sp state.Provider, tp timers.Provider, key, word string, emit func(string, string)) error {
+ s.ElementBag.Add(sp, word)
+ s.MinTime.Add(sp, int64(ts))
+
+ toFire, ok, err := s.TimerTime.Read(sp)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ toFire = int64(mtime.Now().Add(1 * time.Minute))
+ }
+ minTime, _, err := s.MinTime.Read(sp)
+ if err != nil {
+ return err
+ }
+
+ s.OutputState.Set(tp, mtime.Time(toFire).ToTime(), timers.WithOutputTimestamp(mtime.Time(minTime).ToTime()), timers.WithTag(word))
+ s.TimerTime.Write(sp, toFire)
+
+ return nil
+}
+
+type eventtimeSDFStream struct {
Review Comment:
Though TBH we shouldreplace this with a [Periodic Sequence](https://pkg.go.dev/github.com/apache/beam/sdks/v2@v2.47.0-RC1/go/pkg/beam/transforms/periodic#Sequence) which will simplify the example implementation. Replacing ~100 lines with ~5-8 seems like a readability win.
Also, I know that means we can ensure the pipeline terminates.
##########
sdks/go/examples/timer_wordcap/wordcap.go:
##########
@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// timer_wordcap is a toy streaming pipeline that uses State and Timers with PubSub. It
+// does the following:
+//
+// (1) create a topic and publish a few messages to it
+// (2) Set user state and timer
+//
+// NOTE: it only runs on Dataflow and must be manually cancelled.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/state"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
+ "golang.org/x/exp/slog"
+)
+
+var (
+ input = flag.String("input", os.ExpandEnv("$USER-wordcap"), "Pubsub input topic.")
+)
+
+var (
+ data = []string{
+ "foo",
+ "bar",
+ "baz",
+ }
+)
+
+type Stateful struct {
+ ElementBag state.Bag[string]
+ TimerTime state.Value[int64]
+ MinTime state.Combining[int64, int64, int64]
+
+ OutputState timers.ProcessingTime
+}
+
+func NewStateful() *Stateful {
+ return &Stateful{
+ ElementBag: state.MakeBagState[string]("elementBag"),
+ TimerTime: state.MakeValueState[int64]("timerTime"),
+ MinTime: state.MakeCombiningState[int64, int64, int64]("minTiInBag", func(a, b int64) int64 {
+ if a < b {
+ return a
+ }
+ return b
+ }),
+
+ OutputState: timers.InProcessingTime("outputState"),
+ }
+}
+
+func (s *Stateful) OnTimer(ctx context.Context, ts beam.EventTime, tp timers.Provider, key, timerKey, timerTag string, emit func(string, string)) {
+ switch timerKey {
+ case "outputState":
+ log.Infof(ctx, "Timer outputState fired on stateful for element: %v.", key)
+ s.OutputState.Clear(tp)
+ switch timerTag {
+ case "001":
+ log.Infof(ctx, "Timer with tag 001 fired on outputState stateful DoFn.")
+ s.OutputState.Set(tp, mtime.Now().ToTime().Add(1*time.Minute), timers.WithTag(timerTag))
+ emit(timerKey, timerTag)
+ }
+ }
+}
+
+func (s *Stateful) ProcessElement(ctx context.Context, ts beam.EventTime, sp state.Provider, tp timers.Provider, key, word string, emit func(string, string)) error {
+ s.ElementBag.Add(sp, word)
+ s.MinTime.Add(sp, int64(ts))
+
+ toFire, ok, err := s.TimerTime.Read(sp)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ toFire = int64(mtime.Now().Add(1 * time.Minute))
+ }
+ minTime, _, err := s.MinTime.Read(sp)
+ if err != nil {
+ return err
+ }
+
+ s.OutputState.Set(tp, mtime.Time(toFire).ToTime(), timers.WithOutputTimestamp(mtime.Time(minTime).ToTime()), timers.WithTag(word))
Review Comment:
IIRC, we have both minTime and minTime in millis? We should just call https://pkg.go.dev/time#UnixMilli here instead.
Ideally we'd just store time.Times directly in the state, but that feels like a different bug to fix if it doesn't already work.
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
Review Comment:
Do note that you can largely take the existing "Stateful" dofn from the example. There are also other stateful DoFn examples towards the bottom.
Likely don't need that all the "bad" paths are unit tested, other than at least a quick manual check.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1522389858
> @lostluck could you also take a look now?
Ack will do! Thanks!
>
> Regarding the composites, are you proposing a single interface method that would return map[string]TimeDomain? And this method would return just a single KV pair for the built-in timer types (event-time, processing-time) but would be customizable by users in case they want to batch them.
That's right. A single method for the detection interface, and optionally a few reflection based helper functions for the composition bit. Basically it unifies the two current methods by packaging them into a map by default, but are returning the same information: The timer-familyID and its associated domain.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #26101: [Go SDK] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #26101:
URL: https://github.com/apache/beam/pull/26101#issuecomment-1522390165
stop reviewer notifications
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1179682689
##########
sdks/go/pkg/beam/core/graph/fn.go:
##########
@@ -1386,35 +1386,54 @@ func validateState(fn *DoFn, numIn mainInputs) error {
}
func validateOnTimerFn(fn *DoFn) error {
- if _, ok := fn.OnTimerFn(); !ok {
+ if _, ok := fn.methods[onTimerName]; !ok {
err := errors.Errorf("OnTimer function not defined for DoFn: %v", fn.Name())
return errors.SetTopLevelMsgf(err, "OnTimer function not defined for DoFn: %v. Ensure that OnTimer function is implemented for the DoFn.", fn.Name())
}
- return nil
-}
+ if _, ok := fn.methods[onTimerName].TimerProvider(); !ok {
+ err := errors.Errorf("OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v", fn.Name(), fn.PipelineTimers())
+ return errors.SetTopLevelMsgf(err, "OnTimer function doesn't use a TimerProvider, but Timer field is attached to the DoFn(%v): %v"+
+ ", Ensure that you are using the TimerProvider to set and clear the timers.", fn.Name(), fn.PipelineTimers())
+ }
-func validateTimer(fn *DoFn) error {
- if fn.Fn == nil {
- return nil
+ _, otNum, otExists := fn.methods[onTimerName].Emits()
+ _, peNum, peExists := fn.methods[processElementName].Emits()
+
+ if otExists == peExists {
+ if otNum != peNum {
+ return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, no. of emitters used in OnTimer: %v, no. of emitters used in ProcessElement: %v", otNum, peNum)
+ }
+ } else {
+ return fmt.Errorf("OnTimer and ProcessElement functions for DoFn should have exactly same emitters, emitters used in OnTimer: %v, emitters used in ProcessElement: %v", otExists, peExists)
}
+ return nil
+}
+
+func validateTimer(fn *DoFn, numIn mainInputs) error {
pt := fn.PipelineTimers()
- if _, ok := fn.Fn.TimerProvider(); ok {
+ if _, ok := fn.methods[processElementName].TimerProvider(); ok {
+ if numIn == MainSingle {
+ err := errors.Errorf("ProcessElement uses a TimerProvider, but is not keyed")
+ return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but is not keyed. "+
+ "All stateful DoFns must take a key/value pair as an input.")
+ }
if len(pt) == 0 {
err := errors.New("ProcessElement uses a TimerProvider, but no Timer fields are defined in the DoFn")
return errors.SetTopLevelMsgf(err, "ProcessElement uses a TimerProvider, but no timer fields are defined in the DoFn"+
- ", Ensure that you are including the exported timer field in the DoFn that you're using to set/clear timers.")
+ ", Ensure that your DoFn exports the Timer fields used to set and clear timers.")
}
timerKeys := make(map[string]timers.PipelineTimer)
for _, t := range pt {
- k := t.TimerFamily()
- if timer, ok := timerKeys[k]; ok {
- err := errors.Errorf("Duplicate timer key %v", k)
- return errors.SetTopLevelMsgf(err, "Duplicate timer key %v used by %v and %v. Ensure that keys are unique per DoFn", k, timer, t)
- } else {
- timerKeys[k] = t
+ for timerFamilyID := range t.Timers() {
+ if timer, ok := timerKeys[timerFamilyID]; ok {
+ err := errors.Errorf("Duplicate timer key %v", timerFamilyID)
+ return errors.SetTopLevelMsgf(err, "Duplicate timer family ID %v used by %v and %v. Ensure that timer family IDs are unique per DoFn", timerFamilyID, timer, t)
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1179376649
##########
sdks/go/pkg/beam/core/runtime/exec/coder.go:
##########
@@ -145,8 +145,15 @@ func MakeElementEncoder(c *coder.Coder) ElementEncoder {
}
case coder.Timer:
+ tc := coder.SkipW(c).Components[0]
+ if coder.IsKV(tc) {
+ return &timerEncoder{
+ elm: MakeElementEncoder(tc.Components[0]),
+ win: MakeWindowEncoder(c.Window),
+ }
+ }
return &timerEncoder{
- elm: MakeElementEncoder(c.Components[0]),
Review Comment:
Removing the KV check here and always sending key coder from exec/translate
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1179495321
##########
sdks/go/pkg/beam/core/runtime/exec/timers.go:
##########
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "context"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/timers"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+)
+
+// UserTimerAdapter provides a timer provider to be used for manipulating timers.
+type UserTimerAdapter interface {
+ NewTimerProvider(ctx context.Context, manager DataManager, inputTimestamp typex.EventTime, windows []typex.Window, element *MainInput) (timerProvider, error)
+}
+
+type userTimerAdapter struct {
+ sID StreamID
+ timerIDToCoder map[string]*coder.Coder
+}
+
+// NewUserTimerAdapter returns a user timer adapter for the given StreamID and timer coder.
+func NewUserTimerAdapter(sID StreamID, c *coder.Coder, timerCoders map[string]*coder.Coder) UserTimerAdapter {
+ if !coder.IsW(c) {
+ panic(fmt.Sprintf("expected WV coder for user timer %v: %v", sID, c))
+ }
+
+ return &userTimerAdapter{sID: sID, timerIDToCoder: timerCoders}
+}
+
+// NewTimerProvider creates and returns a timer provider to set/clear timers.
+func (u *userTimerAdapter) NewTimerProvider(ctx context.Context, manager DataManager, inputTs typex.EventTime, w []typex.Window, element *MainInput) (timerProvider, error) {
+ userKey := &FullValue{Elm: element.Key.Elm}
+ tp := timerProvider{
+ ctx: ctx,
+ tm: manager,
+ userKey: userKey,
+ inputTimestamp: inputTs,
+ sID: u.sID,
+ window: w,
+ writersByFamily: make(map[string]io.Writer),
+ codersByFamily: u.timerIDToCoder,
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a diff in pull request #26101: [Go SDK] Timers with new datalayer
Posted by "riteshghorse (via GitHub)" <gi...@apache.org>.
riteshghorse commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1177996241
##########
sdks/go/pkg/beam/core/runtime/exec/pardo.go:
##########
@@ -345,6 +355,41 @@ func (n *ParDo) invokeDataFn(ctx context.Context, pn typex.PaneInfo, ws []typex.
return val, nil
}
+func (n *ParDo) InvokeTimerFn(ctx context.Context, fn *funcx.Fn, timerFamilyID string, tmap TimerRecv) (*FullValue, error) {
+ // Defer side input clean-up in case of panic
+ var err error
+ defer func() {
+ if postErr := n.postInvoke(); postErr != nil {
+ err = postErr
+ }
+ }()
+ if err := n.preInvoke(ctx, tmap.Windows, tmap.HoldTimestamp); err != nil {
+ return nil, err
+ }
+
+ var extra []any
+ extra = append(extra, timerFamilyID)
+
+ if tmap.Tag != "" {
+ extra = append(extra, tmap.Tag)
+ }
+ extra = append(extra, n.cache.extra...)
+ val, err := InvokeWithOpts(ctx, fn, tmap.Pane, tmap.Windows, tmap.HoldTimestamp, InvokeOpts{
+ opt: &MainInput{Key: *tmap.Key},
+ bf: n.bf,
+ we: n.we,
+ sa: n.UState,
+ sr: n.reader,
+ ta: n.Timer,
+ tm: n.timerManager,
+ extra: extra,
+ })
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #26101: [WIP] Timers with new datalayer
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #26101:
URL: https://github.com/apache/beam/pull/26101#discussion_r1157544813
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -265,6 +265,30 @@ func (n *DataSource) Process(ctx context.Context) ([]*Checkpoint, error) {
func(bcr *byteCountReader, ptransformID, timerFamilyID string) error {
tmap, err := decodeTimer(cp, wc, bcr)
log.Infof(ctx, "DEBUGLOG: timer received for: %v and %v - %+v err: %v", ptransformID, timerFamilyID, tmap, err)
+ log.Infof(ctx, "OnTimerTransforms = %+v", n.OnTimerTransforms[ptransformID].Fn)
+ if fn, ok := n.OnTimerTransforms[ptransformID].Fn.OnTimerFn(); ok {
+ log.Infof(ctx, "found ontimer method, invoking callback")
+ _, err := n.OnTimerTransforms[ptransformID].InvokeTimerFn(ctx, fn, timerFamilyID, tmap)
+ if err != nil {
+ return errors.WithContext(err, "ontimer callback invocation failed")
+ }
+ }
+ // Check if there's a continuation and return residuals
+ // Needs to be done immediately after processing to not lose the element.
+ // if c := n.getProcessContinuation(); c != nil {
+ // cp, err := n.checkpointThis(ctx, c)
+ // if err != nil {
+ // // Errors during checkpointing should fail a bundle.
+ // return err
+ // }
+ // if cp != nil {
+ // checkpoints = append(checkpoints, cp)
+ // }
+ // }
+ // We've finished processing an element, check if we have finished a split.
+ // if n.incrementIndexAndCheckSplit() {
+ // return splitSuccess
+ // }
Review Comment:
Turns out yes: https://beam.apache.org/blog/looping-timers/
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org