You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "johannaojeling (via GitHub)" <gi...@apache.org> on 2023/02/12 19:10:47 UTC
[GitHub] [beam] johannaojeling opened a new pull request, #25437: [Go SDK]: Allow SDF methods to have context param and error return value
johannaojeling opened a new pull request, #25437:
URL: https://github.com/apache/beam/pull/25437
Addresses #20607 by allowing SDF methods to accept an optional context parameter and return an optional error. The updated signatures are:
- CreateInitialRestriction(context.Context?, element) (restriction, error?)
- SplitRestriction(context.Context?, element, restriction) ([]restriction, error?)
- RestrictionSize(context.Context?, element, restriction) (float64, error?)
- CreateTracker(context.Context?, restriction) (sdf.RTracker, error?)
- TruncateRestriction(context.Context?, sdf.RTracker, restriction) (restriction, error?)
------------------------
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] lostluck merged pull request #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck merged PR #25437:
URL: https://github.com/apache/beam/pull/25437
--
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] codecov[bot] commented on pull request #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #25437:
URL: https://github.com/apache/beam/pull/25437#issuecomment-1427110498
# [Codecov](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#25437](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b6665c0) into [master](https://codecov.io/gh/apache/beam/commit/198b93eacef5bae6e3835e73674d6b46b3492fcd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (198b93e) will **decrease** coverage by `0.11%`.
> The diff coverage is `54.23%`.
```diff
@@ Coverage Diff @@
## master #25437 +/- ##
==========================================
- Coverage 72.86% 72.75% -0.11%
==========================================
Files 750 751 +1
Lines 99501 99760 +259
==========================================
+ Hits 72497 72583 +86
- Misses 25636 25787 +151
- Partials 1368 1390 +22
```
| Flag | Coverage Δ | |
|---|---|---|
| go | `51.73% <54.23%> (-0.16%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...o/pkg/beam/core/runtime/exec/sdf\_invokers\_arity.go](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9zZGZfaW52b2tlcnNfYXJpdHkuZ28=) | `30.61% <30.61%> (ø)` | |
| [sdks/go/pkg/beam/core/runtime/exec/sdf.go](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9zZGYuZ28=) | `66.12% <49.38%> (-4.89%)` | :arrow_down: |
| [sdks/go/pkg/beam/core/runtime/exec/sdf\_invokers.go](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9zZGZfaW52b2tlcnMuZ28=) | `76.30% <86.53%> (+6.12%)` | :arrow_up: |
| [sdks/go/pkg/beam/core/graph/fn.go](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2ZuLmdv) | `85.11% <96.77%> (+0.43%)` | :arrow_up: |
| [sdks/go/pkg/beam/core/runtime/exec/datasource.go](https://codecov.io/gh/apache/beam/pull/25437?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9kYXRhc291cmNlLmdv) | `68.93% <100.00%> (ø)` | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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 #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #25437:
URL: https://github.com/apache/beam/pull/25437#issuecomment-1432382838
Work got away from me, preventing me from spending time on this. First thing I do tomorrow though.
--
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] johannaojeling commented on a diff in pull request #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "johannaojeling (via GitHub)" <gi...@apache.org>.
johannaojeling commented on code in PR #25437:
URL: https://github.com/apache/beam/pull/25437#discussion_r1108977986
##########
sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:
##########
@@ -39,61 +43,45 @@ import (
// cirInvoker is an invoker for CreateInitialRestriction.
type cirInvoker struct {
- fn *funcx.Fn
- args []any // Cache to avoid allocating new slices per-element.
- call func(elms *FullValue) (rest any)
+ fn *funcx.Fn
+ args []any // Cache to avoid allocating new slices per-element.
+ ctxIdx int
+ call func() (rest any, err error)
}
func newCreateInitialRestrictionInvoker(fn *funcx.Fn) (*cirInvoker, error) {
n := &cirInvoker{
fn: fn,
args: make([]any, len(fn.Param)),
}
+
+ var ok bool
+ if n.ctxIdx, ok = fn.Context(); !ok {
+ n.ctxIdx = -1
+ }
+
if err := n.initCallFn(); err != nil {
return nil, errors.WithContext(err, "sdf CreateInitialRestriction invoker")
}
return n, nil
}
-func (n *cirInvoker) initCallFn() error {
- // Expects a signature of the form:
- // (key?, value) restriction
- // TODO(BEAM-9643): Link to full documentation.
- switch fnT := n.fn.Fn.(type) {
- case reflectx.Func1x1:
Review Comment:
Ok I see. Thanks for the explanation and review!
--
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] johannaojeling commented on pull request #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "johannaojeling (via GitHub)" <gi...@apache.org>.
johannaojeling commented on PR #25437:
URL: https://github.com/apache/beam/pull/25437#issuecomment-1427115132
@lostluck I tried running with your new prism runner. Very nice!
--
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 #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #25437:
URL: https://github.com/apache/beam/pull/25437#issuecomment-1430798794
I am intending to get to this PR properly tomorrow.
--
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] johannaojeling commented on a diff in pull request #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "johannaojeling (via GitHub)" <gi...@apache.org>.
johannaojeling commented on code in PR #25437:
URL: https://github.com/apache/beam/pull/25437#discussion_r1107656944
##########
sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:
##########
@@ -39,61 +43,45 @@ import (
// cirInvoker is an invoker for CreateInitialRestriction.
type cirInvoker struct {
- fn *funcx.Fn
- args []any // Cache to avoid allocating new slices per-element.
- call func(elms *FullValue) (rest any)
+ fn *funcx.Fn
+ args []any // Cache to avoid allocating new slices per-element.
+ ctxIdx int
+ call func() (rest any, err error)
}
func newCreateInitialRestrictionInvoker(fn *funcx.Fn) (*cirInvoker, error) {
n := &cirInvoker{
fn: fn,
args: make([]any, len(fn.Param)),
}
+
+ var ok bool
+ if n.ctxIdx, ok = fn.Context(); !ok {
+ n.ctxIdx = -1
+ }
+
if err := n.initCallFn(); err != nil {
return nil, errors.WithContext(err, "sdf CreateInitialRestriction invoker")
}
return n, nil
}
-func (n *cirInvoker) initCallFn() error {
- // Expects a signature of the form:
- // (key?, value) restriction
- // TODO(BEAM-9643): Link to full documentation.
- switch fnT := n.fn.Fn.(type) {
- case reflectx.Func1x1:
Review Comment:
Thanks, hm okay, not sure if I fully understand the preferred way forward. Do you mean a) we should strive to handle the invocation of SDF methods with [exec.Invoke](https://github.com/apache/beam/blob/fd2b16139fd03d3221975ca88fa516f7ad41b6a4/sdks/go/pkg/beam/core/runtime/exec/fn.go#L73) and remove the individual SDF invokers, b) there is no need to check the reflectx.FuncMxN cases for SDF methods but they should always be invoked with the default case `n.fn.Fn.Call()`, or c) something else? 🙂
I don't have the full picture but what I've seen so far when looking at the invocation of ProcessElement, is that the reflectx.FuncMxN cases match when there has been a generic registration of the DoFn. Currently there seems to be support for wrapping the standard methods Setup, StartBundle, ProcessElement, FinishBundle and Teardown into these types, but not for SDF methods, so for SDF methods the reflectx.FuncMxN cases won't match and the default case will happen. There could be other things going on that I'm unaware of though, so I don't know how valid this observation is.
--
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 #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #25437:
URL: https://github.com/apache/beam/pull/25437#discussion_r1108967075
##########
sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:
##########
@@ -39,61 +43,45 @@ import (
// cirInvoker is an invoker for CreateInitialRestriction.
type cirInvoker struct {
- fn *funcx.Fn
- args []any // Cache to avoid allocating new slices per-element.
- call func(elms *FullValue) (rest any)
+ fn *funcx.Fn
+ args []any // Cache to avoid allocating new slices per-element.
+ ctxIdx int
+ call func() (rest any, err error)
}
func newCreateInitialRestrictionInvoker(fn *funcx.Fn) (*cirInvoker, error) {
n := &cirInvoker{
fn: fn,
args: make([]any, len(fn.Param)),
}
+
+ var ok bool
+ if n.ctxIdx, ok = fn.Context(); !ok {
+ n.ctxIdx = -1
+ }
+
if err := n.initCallFn(); err != nil {
return nil, errors.WithContext(err, "sdf CreateInitialRestriction invoker")
}
return n, nil
}
-func (n *cirInvoker) initCallFn() error {
- // Expects a signature of the form:
- // (key?, value) restriction
- // TODO(BEAM-9643): Link to full documentation.
- switch fnT := n.fn.Fn.(type) {
- case reflectx.Func1x1:
Review Comment:
OK, now that I've looked at the whole CL I understand your question. I think.
You're right that the generic `register` package doesn't currently wrap these methods in `CallerMxN` type things. Some of the scaffolding here is forward looking in the event we augment the `register` package to do this. I think the old code generator also did some of this, but it wasn't ported to the `register` package yet.
At present, the only methods that get wrapped for sure are the DoFn Bundle Lifecycle methods (StartBundle, ProcessElement, FinishBundle). The reason for this is those methods are called significantly more often, so paying the repeated overhead of reflection for every call would be very noticible on profiles.
Conversely, the SDF methods are called rarely. Some only meaningfully happen on Dynamic Splitting, at the runner's request. So the overhead is unlikely to appear in any execution profile, which is the motivation for all of these wrappings and type assertions in the first place.
----
Basically as much as the current set up is neat and versatile, TBH I'd love to have an alternate DoFn API that is much simpler for users by being more opinionated, and simpler to invoke execution side.
At one extreme version we lose the ability to use arbitrary functions as DoFns, but it would be much clearer to everyone what's going on, and how we execute things at run time. Mentally, I've been calling this the "performance" API.
--
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 #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #25437:
URL: https://github.com/apache/beam/pull/25437#discussion_r1106682555
##########
sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:
##########
@@ -39,61 +43,45 @@ import (
// cirInvoker is an invoker for CreateInitialRestriction.
type cirInvoker struct {
- fn *funcx.Fn
- args []any // Cache to avoid allocating new slices per-element.
- call func(elms *FullValue) (rest any)
+ fn *funcx.Fn
+ args []any // Cache to avoid allocating new slices per-element.
+ ctxIdx int
+ call func() (rest any, err error)
}
func newCreateInitialRestrictionInvoker(fn *funcx.Fn) (*cirInvoker, error) {
n := &cirInvoker{
fn: fn,
args: make([]any, len(fn.Param)),
}
+
+ var ok bool
+ if n.ctxIdx, ok = fn.Context(); !ok {
+ n.ctxIdx = -1
+ }
+
if err := n.initCallFn(); err != nil {
return nil, errors.WithContext(err, "sdf CreateInitialRestriction invoker")
}
return n, nil
}
-func (n *cirInvoker) initCallFn() error {
- // Expects a signature of the form:
- // (key?, value) restriction
- // TODO(BEAM-9643): Link to full documentation.
- switch fnT := n.fn.Fn.(type) {
- case reflectx.Func1x1:
Review Comment:
Up until this point, the SDF methods all had known "arities" of Input Parameters and output parameters. So, the normal Beam DoFn Bundle Lifecycle methods (StartBundle, ProcessElement, FinishBundle) could have fairly arbitrary inputs and up to 5 outputs (ignoring emitters).
The idea behind splitting out the SDF invokers was because they had a limited set of "arities" they could have, so they could take a shorter less abstract path. The main trick is they aren't exactly the same as the other methods (dealing with emitters, iterators, etc).
There's no great reason to split them out now if the general handling works for them though.
--
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 #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #25437:
URL: https://github.com/apache/beam/pull/25437#issuecomment-1430798381
> @lostluck I tried running with your new prism runner. Very nice!
Delightful! It's better than the Go Direct Runner, but it's still got a long way to go before it's where I want it. Soon! :D
--
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] johannaojeling commented on a diff in pull request #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "johannaojeling (via GitHub)" <gi...@apache.org>.
johannaojeling commented on code in PR #25437:
URL: https://github.com/apache/beam/pull/25437#discussion_r1103864114
##########
sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:
##########
@@ -39,61 +43,45 @@ import (
// cirInvoker is an invoker for CreateInitialRestriction.
type cirInvoker struct {
- fn *funcx.Fn
- args []any // Cache to avoid allocating new slices per-element.
- call func(elms *FullValue) (rest any)
+ fn *funcx.Fn
+ args []any // Cache to avoid allocating new slices per-element.
+ ctxIdx int
+ call func() (rest any, err error)
}
func newCreateInitialRestrictionInvoker(fn *funcx.Fn) (*cirInvoker, error) {
n := &cirInvoker{
fn: fn,
args: make([]any, len(fn.Param)),
}
+
+ var ok bool
+ if n.ctxIdx, ok = fn.Context(); !ok {
+ n.ctxIdx = -1
+ }
+
if err := n.initCallFn(); err != nil {
return nil, errors.WithContext(err, "sdf CreateInitialRestriction invoker")
}
return n, nil
}
-func (n *cirInvoker) initCallFn() error {
- // Expects a signature of the form:
- // (key?, value) restriction
- // TODO(BEAM-9643): Link to full documentation.
- switch fnT := n.fn.Fn.(type) {
- case reflectx.Func1x1:
Review Comment:
I'm trying to understand in which cases the SDF methods are of type `reflectx.FuncMxN` (now moved to the generated file). I see the standard DoFn methods are wrapped into these types when using generic registration, however I cannot find that the same applies to the SDF methods
--
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 #25437: [Go SDK]: Allow SDF methods to have context param and error return value
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25437:
URL: https://github.com/apache/beam/pull/25437#issuecomment-1427114027
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @lostluck 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