You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/04/21 15:23:36 UTC
[GitHub] [beam] damccorm opened a new pull request, #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
damccorm opened a new pull request, #17429:
URL: https://github.com/apache/beam/pull/17429
This is the first step towards allowing users to fully replace the code generator with a DoFn registration function (design doc https://docs.google.com/document/d/1MLZLUhb-Sn7AJ666yn3TnYgMKE6wiXIBs7cjB_20hMA/edit?usp=sharing). Generally, this accomplishes this by:
- Updating to Go 1.18
- Adding registration.tmpl to generate registration.go with a bunch of arity specific registration functions. At the moment, these register struct wrappers, types, and funcs for the passed in DoFn.
- Update specialize/main.go to provide some functions used by registration.tmpl
Because the tmpl file is hard to review in isolation, I would do so with the generated file side by side so that you can see how the template translates into the final code.
**Next steps**
After this is in, I will follow up with prs to:
1. Handle emitter/iter registration
2. Handle combiner registration
3. Perf testing
4. Updates to the docs/our examples
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [x] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [x] 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/#make-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)
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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863830365
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
+ {{$lowerName}}In := -1
+ {{$lowerName}}Out := -1
+ {{$lowerName}}Method := reflect.ValueOf(doFn).MethodByName("{{$upperName}}")
+ if {{$lowerName}}Method.IsValid() {
+ {{$lowerName}}In = {{$lowerName}}Method.Type().NumIn()
+ {{$lowerName}}Out = {{$lowerName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto $startFinishBundleMaxIn}}
+ case {{$lowerName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $numParams)}}{{if $possibleCombos}}
+ case {{$lowerName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})(nil)).Elem(), func(fn interface{}) reflectx.Func { return &caller{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}{fn: fn.(func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})} })
+ {{$funcRegister := (makeStructRegisterEntry $lowerName $upperName $funcCombo (list))}}{{if $funcOut}}{{$funcRegister = (makeStructRegisterEntry $lowerName $upperName $funcCombo (list "error"))}}{{end}}
+ return func(fn interface{}) reflectx.Func { return {{$funcRegister}} }
+ } {{end}}{{end}}{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+ }
+ return nil
+}
+{{end}}{{end}}{{define "BuildWrapper_SetupTeardown"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "setup")}}{{$lowerName = "setup"}}{{$upperName = "Setup"}}{{end}}{{if (eq .func "teardown")}}{{$lowerName = "teardown"}}{{$upperName = "Teardown"}}{{end}}
+func build{{$upperName}}Wrapper(doFn interface{}) func(interface{}) reflectx.Func {
+ if _, ok := doFn.({{$lowerName}}0x0); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func())
+ return &caller0x0{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.({{$lowerName}}0x0).{{$upperName}}()
+ })
+ }
+ } else if _, ok := doFn.({{$lowerName}}1x0); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context))
+ return &caller1x0[context.Context]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context))(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) {
+ fn.({{$lowerName}}1x0).{{$upperName}}(a0)
+ })
+ }
+ } else if _, ok := doFn.({{$lowerName}}0x1); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func() error)
+ return &caller0x1[error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func() error)(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() error {
+ return fn.({{$lowerName}}0x1).{{$upperName}}()
+ })
+ }
+ } else if _, ok := doFn.({{$lowerName}}1x1); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context) error)
+ return &caller1x1[context.Context, error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context) error)(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) error {
+ return fn.({{$lowerName}}1x1).{{$upperName}}(a0)
+ })
+ }
+ }
+
+ return nil
+}
+{{end}}
+// 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.
+
+// Code generated from registration.tmpl. DO NOT EDIT.
+
+package beam
Review Comment:
Oop - good catch!
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863219458
##########
.github/workflows/go_tests.yml:
##########
@@ -39,7 +39,7 @@ jobs:
fetch-depth: 2
- uses: actions/setup-go@v2
with:
- go-version: '1.16'
+ go-version: '1.18'
Review Comment:
I'm guessing there's no easy/good way to have a constants/ENV file for github actions, that we can share with everything so we can update this in a single place?
##########
sdks/go/cmd/specialize/main.go:
##########
@@ -238,3 +246,130 @@ func upto(i int) []int {
}
return ret
}
+
+func add(i int, j int) int {
+ return i + j
+}
+
+func mult(i int, j int) int {
+ return i * j
+}
+
+func dict(values ...interface{}) map[string]interface{} {
+ dict := make(map[string]interface{}, len(values)/2)
+ if len(values)%2 != 0 {
+ panic("Invalid dictionary call")
+ }
+ for i := 0; i < len(values); i += 2 {
+ dict[values[i].(string)] = values[i+1]
+ }
+
+ return dict
+}
+
+func list(values ...string) []string {
+ return values
+}
+
+func genericTypingRepresentation(in int, out int, includeType bool) string {
+ seenElements := false
+ typing := ""
+ if in > 0 {
+ typing += fmt.Sprintf("[I%v", 0)
+ for i := 1; i < in; i++ {
+ typing += fmt.Sprintf(", I%v", i)
+ }
+ seenElements = true
+ }
+ if out > 0 {
+ i := 0
+ if !seenElements {
+ typing += fmt.Sprintf("[R%v", 0)
+ i++
+ }
+ for i < out {
+ typing += fmt.Sprintf(", R%v", i)
+ i++
+ }
+ seenElements = true
+ }
+
+ if seenElements {
+ if includeType {
+ typing += " any"
+ }
+ typing += "]"
+ }
+
+ return typing
+}
+
+func possibleBundleLifecycleParameterCombos(numInInterface interface{}, processElementInInterface interface{}) [][]string {
+ numIn := numInInterface.(int)
+ processElementIn := processElementInInterface.(int)
+ ordered_known_parameter_options := []string{"context.Context", "typex.PaneInfo", "[]typex.Window", "typex.EventTime", "typex.BundleFinalization"}
+ // Because of how Bundle lifecycle functions are invoked, all known parameters must preced unknown options and be in order.
Review Comment:
```suggestion
// Because of how Bundle lifecycle functions are invoked, all known parameters must preceed unknown options and be in order.
```
##########
sdks/go/pkg/beam/registration/registration_test.go:
##########
@@ -0,0 +1,218 @@
+// 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.
+
+//go:generate go install github.com/apache/beam/sdks/v2/go/cmd/specialize
+//go:generate specialize --package=beam --input=registration.tmpl --x=data,universals --imports=typex
+//go:generate go fmt
Review Comment:
I'd prefer if this weren't in a test file, but I think we can move it to a `doc.go` file, and that way, we can have a proper package comment explaining how it works.
Also `--package=registration`, not `beam`.
We could also have a full file testable example on the side to demonstrate the packages use.
https://go.dev/blog/examples#:~:text=Larger%20examples
##########
sdks/go/pkg/beam/registration/registration_test.go:
##########
@@ -0,0 +1,218 @@
+// 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.
+
+//go:generate go install github.com/apache/beam/sdks/v2/go/cmd/specialize
+//go:generate specialize --package=beam --input=registration.tmpl --x=data,universals --imports=typex
+//go:generate go fmt
+
+package beam
Review Comment:
```suggestion
package registration
```
##########
sdks/go/cmd/specialize/main.go:
##########
@@ -238,3 +246,130 @@ func upto(i int) []int {
}
return ret
}
+
+func add(i int, j int) int {
+ return i + j
+}
+
+func mult(i int, j int) int {
+ return i * j
+}
+
+func dict(values ...interface{}) map[string]interface{} {
+ dict := make(map[string]interface{}, len(values)/2)
+ if len(values)%2 != 0 {
+ panic("Invalid dictionary call")
+ }
+ for i := 0; i < len(values); i += 2 {
+ dict[values[i].(string)] = values[i+1]
+ }
+
+ return dict
+}
+
+func list(values ...string) []string {
+ return values
+}
+
+func genericTypingRepresentation(in int, out int, includeType bool) string {
+ seenElements := false
+ typing := ""
+ if in > 0 {
+ typing += fmt.Sprintf("[I%v", 0)
+ for i := 1; i < in; i++ {
+ typing += fmt.Sprintf(", I%v", i)
+ }
+ seenElements = true
+ }
+ if out > 0 {
+ i := 0
+ if !seenElements {
+ typing += fmt.Sprintf("[R%v", 0)
+ i++
+ }
+ for i < out {
+ typing += fmt.Sprintf(", R%v", i)
+ i++
+ }
+ seenElements = true
+ }
+
+ if seenElements {
+ if includeType {
+ typing += " any"
+ }
+ typing += "]"
+ }
+
+ return typing
+}
+
+func possibleBundleLifecycleParameterCombos(numInInterface interface{}, processElementInInterface interface{}) [][]string {
Review Comment:
I would recommend adding a test to this package/binary for this. It's allowed to add tests to `package main`,
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
+ {{$lowerName}}In := -1
+ {{$lowerName}}Out := -1
+ {{$lowerName}}Method := reflect.ValueOf(doFn).MethodByName("{{$upperName}}")
+ if {{$lowerName}}Method.IsValid() {
+ {{$lowerName}}In = {{$lowerName}}Method.Type().NumIn()
+ {{$lowerName}}Out = {{$lowerName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto $startFinishBundleMaxIn}}
+ case {{$lowerName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $numParams)}}{{if $possibleCombos}}
+ case {{$lowerName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})(nil)).Elem(), func(fn interface{}) reflectx.Func { return &caller{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}{fn: fn.(func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})} })
+ {{$funcRegister := (makeStructRegisterEntry $lowerName $upperName $funcCombo (list))}}{{if $funcOut}}{{$funcRegister = (makeStructRegisterEntry $lowerName $upperName $funcCombo (list "error"))}}{{end}}
+ return func(fn interface{}) reflectx.Func { return {{$funcRegister}} }
Review Comment:
This block here that wraps to a caller generation, feels like an opportunity to reduce lines. If we add one further layer of generic helper (0x0 though 7x1) for the 'caller' generation and wrapping we could save 6 lines per invocation.
Defining them should be 10 lines each including the blank line spacer, with 14 helpers shared among Start and Finish bundles, which happens around ~2600 times, that's new 140 lines, but still around 16k lines saved or so.
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
+ {{$lowerName}}In := -1
+ {{$lowerName}}Out := -1
+ {{$lowerName}}Method := reflect.ValueOf(doFn).MethodByName("{{$upperName}}")
+ if {{$lowerName}}Method.IsValid() {
+ {{$lowerName}}In = {{$lowerName}}Method.Type().NumIn()
+ {{$lowerName}}Out = {{$lowerName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto $startFinishBundleMaxIn}}
+ case {{$lowerName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $numParams)}}{{if $possibleCombos}}
+ case {{$lowerName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})(nil)).Elem(), func(fn interface{}) reflectx.Func { return &caller{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}{fn: fn.(func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})} })
+ {{$funcRegister := (makeStructRegisterEntry $lowerName $upperName $funcCombo (list))}}{{if $funcOut}}{{$funcRegister = (makeStructRegisterEntry $lowerName $upperName $funcCombo (list "error"))}}{{end}}
+ return func(fn interface{}) reflectx.Func { return {{$funcRegister}} }
+ } {{end}}{{end}}{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+ }
+ return nil
+}
+{{end}}{{end}}{{define "BuildWrapper_SetupTeardown"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "setup")}}{{$lowerName = "setup"}}{{$upperName = "Setup"}}{{end}}{{if (eq .func "teardown")}}{{$lowerName = "teardown"}}{{$upperName = "Teardown"}}{{end}}
+func build{{$upperName}}Wrapper(doFn interface{}) func(interface{}) reflectx.Func {
+ if _, ok := doFn.({{$lowerName}}0x0); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func())
+ return &caller0x0{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.({{$lowerName}}0x0).{{$upperName}}()
+ })
+ }
+ } else if _, ok := doFn.({{$lowerName}}1x0); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context))
+ return &caller1x0[context.Context]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context))(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) {
+ fn.({{$lowerName}}1x0).{{$upperName}}(a0)
+ })
+ }
+ } else if _, ok := doFn.({{$lowerName}}0x1); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func() error)
+ return &caller0x1[error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func() error)(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() error {
+ return fn.({{$lowerName}}0x1).{{$upperName}}()
+ })
+ }
+ } else if _, ok := doFn.({{$lowerName}}1x1); ok {
+ {{$lowerName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context) error)
+ return &caller1x1[context.Context, error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context) error)(nil)).Elem(), {{$lowerName}}Caller)
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) error {
+ return fn.({{$lowerName}}1x1).{{$upperName}}(a0)
+ })
+ }
+ }
+
+ return nil
+}
+{{end}}
+// 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.
+
+// Code generated from registration.tmpl. DO NOT EDIT.
+
+package beam
Review Comment:
```suggestion
package registration
```
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
Review Comment:
The vast bulk of the generated file is handling startbundle and finish bundle. I'm curious if we can somehow modify what we have as `buildStartBundleWrapper9`, passing in the parameter cap, and populate the unused types with `int` or `struct{}`?
(The raw file is viewable at least, which is nice https://raw.githubusercontent.com/apache/beam/9490267e52bb8f50a829f8c683db0beebc0b089a/sdks/go/pkg/beam/registration/registration.go)
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1116501967
Run Python PreCommit
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1108585124
> but how do you feel about merging this after the 2.39 cut, so the first release that requires Go 1.18 is 2.40?
That's fine by me - is your goal just to get some additional validation with all of us building stuff on Go 1.18 for a while?
> Whenever we merge it though, this PR must also include a note in CHANGES.md for whichever version it affects
Yeah, I was planning on adding something in CHANGES.md in a follow up pr (along with doc changes) once all the generics work is in - I can also add a placeholder in this PR once 2.39 is cut (and the 2.40 release notes are ready) 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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1105518004
Run SQL_Java17 PreCommit
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1116444800
Run Python_PVR_Flink PreCommit
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1105760936
R: @lostluck
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1106974506
I don't yet have time to look at this just yet, but how do you feel about merging this *after* the 2.39 cut, so the first release that requires Go 1.18 is 2.40?
Whenever we merge it though, this PR must also include a note in CHANGES.md for whichever version it affects, since the Go SDK would then require a minimum Go version of 1.18. The minimum doesn't just apply to the generic using packages, but to the entire SDK.
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1110318446
LOL: "-45%" test coverage.
+293,588 lines.
Hmmm. When I look at this properly, I'll see if we can't figure out other helper functions to bring that down. I think that's larger than the rest of the SDK...
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1110343929
> Hmmm. When I look at this properly, I'll see if we can't figure out other helper functions to bring that down. I think that's larger than the rest of the SDK...
Actually, rethinking this I think I have an idea of how I can shrink this that I'd originally missed - I'll take another pass at doing that in the morning. In general, its a little messy because I still need to hold on to types, but it should help significantly. Feel free to hold off on reviewing until then
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1116501780
Run Portable_Python PreCommit
--
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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r864441839
##########
sdks/go/cmd/specialize/main_test.go:
##########
@@ -0,0 +1,146 @@
+// 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 main
+
+import (
+ "fmt"
+ "testing"
+)
+
+func TestGenericTypingRepresentation(t *testing.T) {
+ tests := []struct {
+ in int
+ out int
+ includeType bool
+ representation string
+ }{
+ {0, 0, true, ""},
+ {0, 0, false, ""},
+ {0, 1, true, "[R0 any]"},
+ {0, 1, false, "[R0]"},
+ {0, 3, true, "[R0, R1, R2 any]"},
+ {0, 3, false, "[R0, R1, R2]"},
+ {1, 0, true, "[I0 any]"},
+ {1, 0, false, "[I0]"},
+ {3, 0, true, "[I0, I1, I2 any]"},
+ {3, 0, false, "[I0, I1, I2]"},
+ {1, 1, true, "[I0, R0 any]"},
+ {1, 1, false, "[I0, R0]"},
+ {4, 1, true, "[I0, I1, I2, I3, R0 any]"},
+ {4, 1, false, "[I0, I1, I2, I3, R0]"},
+ {4, 2, true, "[I0, I1, I2, I3, R0, R1 any]"},
+ {4, 2, false, "[I0, I1, I2, I3, R0, R1]"},
+ {1, 2, true, "[I0, R0, R1 any]"},
+ {1, 2, false, "[I0, R0, R1]"},
+ }
+
+ for _, test := range tests {
+ if got, want := genericTypingRepresentation(test.in, test.out, test.includeType), test.representation; got != want {
+ t.Errorf("genericTypingRepresentation(%v, %v, %v) = %v, want: %v", test.in, test.out, test.includeType, got, want)
+ }
+ }
+}
+
+func TestPossibleBundleLifecycleParameterCombos(t *testing.T) {
+ tests := []struct {
+ numIn int
+ processElementIn int
+ representation [][]string
+ }{
+ {0, 0, [][]string{[]string{}}},
+ {0, 1, [][]string{[]string{}}},
+ {1, 0, [][]string{[]string{"context.Context"}, []string{"typex.PaneInfo"}, []string{"[]typex.Window"}, []string{"typex.EventTime"}, []string{"typex.BundleFinalization"}}},
+ {1, 1, [][]string{[]string{"I0"}, []string{"context.Context"}, []string{"typex.PaneInfo"}, []string{"[]typex.Window"}, []string{"typex.EventTime"}, []string{"typex.BundleFinalization"}}},
+ {2, 1, [][]string{[]string{"context.Context", "I0"}, []string{"context.Context", "typex.PaneInfo"}, []string{"context.Context", "[]typex.Window"}, []string{"context.Context", "typex.EventTime"}, []string{"context.Context", "typex.BundleFinalization"},
+ []string{"typex.PaneInfo", "I0"}, []string{"typex.PaneInfo", "[]typex.Window"}, []string{"typex.PaneInfo", "typex.EventTime"}, []string{"typex.PaneInfo", "typex.BundleFinalization"},
+ []string{"[]typex.Window", "I0"}, []string{"[]typex.Window", "typex.EventTime"}, []string{"[]typex.Window", "typex.BundleFinalization"},
+ []string{"typex.EventTime", "I0"}, []string{"typex.EventTime", "typex.BundleFinalization"},
+ []string{"typex.BundleFinalization", "I0"}}},
+ {2, 2, [][]string{[]string{"context.Context", "I1"}, []string{"context.Context", "typex.PaneInfo"}, []string{"context.Context", "[]typex.Window"}, []string{"context.Context", "typex.EventTime"}, []string{"context.Context", "typex.BundleFinalization"},
+ []string{"typex.PaneInfo", "I1"}, []string{"typex.PaneInfo", "[]typex.Window"}, []string{"typex.PaneInfo", "typex.EventTime"}, []string{"typex.PaneInfo", "typex.BundleFinalization"},
+ []string{"[]typex.Window", "I1"}, []string{"[]typex.Window", "typex.EventTime"}, []string{"[]typex.Window", "typex.BundleFinalization"},
+ []string{"typex.EventTime", "I1"}, []string{"typex.EventTime", "typex.BundleFinalization"},
+ []string{"typex.BundleFinalization", "I1"},
+ []string{"I0", "I1"}}},
+ {2, 3, [][]string{[]string{"context.Context", "I2"}, []string{"context.Context", "typex.PaneInfo"}, []string{"context.Context", "[]typex.Window"}, []string{"context.Context", "typex.EventTime"}, []string{"context.Context", "typex.BundleFinalization"},
+ []string{"typex.PaneInfo", "I2"}, []string{"typex.PaneInfo", "[]typex.Window"}, []string{"typex.PaneInfo", "typex.EventTime"}, []string{"typex.PaneInfo", "typex.BundleFinalization"},
+ []string{"[]typex.Window", "I2"}, []string{"[]typex.Window", "typex.EventTime"}, []string{"[]typex.Window", "typex.BundleFinalization"},
+ []string{"typex.EventTime", "I2"}, []string{"typex.EventTime", "typex.BundleFinalization"},
+ []string{"typex.BundleFinalization", "I2"},
+ []string{"I1", "I2"}}},
+ }
+
+ for _, test := range tests {
+ got, want := possibleBundleLifecycleParameterCombos(test.numIn, test.processElementIn), test.representation
+ assertRepresentationEquals(t, got, want, fmt.Sprintf("possibleBundleLifecycleParameterCombos(%v, %v)", test.numIn, test.processElementIn))
+ }
+}
+
+func TestMakeStructRegisterEntry(t *testing.T) {
+ tests := []struct {
+ structName string
+ functionName string
+ ins []string
+ outs []string
+ entry string
+ }{
+ {"StartBundle", "startBundle", []string{}, []string{}, "reflectx.MakeFunc(func() { fn.(StartBundle0x0).startBundle() })"},
+ {"StartBundle", "startBundle", []string{"context.Context"}, []string{}, "reflectx.MakeFunc(func(a0 context.Context) { fn.(StartBundle1x0[context.Context]).startBundle(a0) })"},
+ {"StartBundle", "startBundle", []string{"context.Context", "I0"}, []string{}, "reflectx.MakeFunc(func(a0 context.Context, a1 I0) { fn.(StartBundle2x0[context.Context, I0]).startBundle(a0, a1) })"},
+ {"FinishBundle", "finishBundle", []string{}, []string{"[]typex.Window"}, "reflectx.MakeFunc(func() ([]typex.Window){ return fn.(FinishBundle0x1[[]typex.Window]).finishBundle() })"},
+ {"FinishBundle", "finishBundle", []string{}, []string{"[]typex.Window", "R0"}, "reflectx.MakeFunc(func() ([]typex.Window, R0){ return fn.(FinishBundle0x2[[]typex.Window, R0]).finishBundle() })"},
+ {"FinishBundle", "finishBundle", []string{"I0"}, []string{"[]typex.Window", "R0"}, "reflectx.MakeFunc(func(a0 I0) ([]typex.Window, R0){ return fn.(FinishBundle1x2[I0, []typex.Window, R0]).finishBundle(a0) })"},
+ {"FinishBundle", "finishBundle", []string{"context.Context", "I0"}, []string{"[]typex.Window", "R0"}, "reflectx.MakeFunc(func(a0 context.Context, a1 I0) ([]typex.Window, R0){ return fn.(FinishBundle2x2[context.Context, I0, []typex.Window, R0]).finishBundle(a0, a1) })"},
+ }
+
+ for _, test := range tests {
+ if got, want := makeStructRegisterEntry(test.structName, test.functionName, test.ins, test.outs), test.entry; got != want {
+ t.Errorf("makeStructRegisterEntry(\"%v\", \"%v\", %v, %v) = %v, want: %v", test.structName, test.functionName, test.ins, test.outs, got, want)
+ }
+ }
+}
+
+func assertRepresentationEquals(t *testing.T, got [][]string, want [][]string, functionInvocation string) {
Review Comment:
Yep, simplifying a little bit makes moving into the caller palatable
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck merged PR #17429:
URL: https://github.com/apache/beam/pull/17429
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1118969154
> This is great! One last thing that needs to be in here:
> CHANGES.md regarding moving the minimum Go version to 1.18
Yep, thanks for the reminder - I'd been waiting until the 2.39 cut for this. Updated!
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1108870404
> > but how do you feel about merging this after the 2.39 cut, so the first release that requires Go 1.18 is 2.40?
>
> Another option (which I might like even more) is to review this PR as is, and then undo all the changes except for the template file and specialize/main.go. That would allow me to keep putting up my remaining generics changes for review, so that when we reach 2.40 all that needs to happen is running `go generate`, go.mod and CI changes, and adding back `registration_tests.go`.
>
> That would save me some hard context shifts, but it may be more trouble than its worth - regardless, I'd prefer we prioritize finishing the [watermark reviews](https://github.com/apache/beam/pull/17374), and by the time those are all through 2.39 may have been cut (or be close to being cut)
I like this idea. It's a happy medium, and will make later improvements easier to review. My goal is to prioritize the Watermark and checkpoint reviews once my oncall is handed off.
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1105433948
# [Codecov](https://codecov.io/gh/apache/beam/pull/17429?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 [#17429](https://codecov.io/gh/apache/beam/pull/17429?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (3287c3c) into [master](https://codecov.io/gh/apache/beam/commit/73c683fe2b0a02ae7a5327e0cdf2991079ca3500?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (73c683f) will **decrease** coverage by `0.01%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #17429 +/- ##
==========================================
- Coverage 73.76% 73.75% -0.02%
==========================================
Files 686 687 +1
Lines 90117 90330 +213
==========================================
+ Hits 66477 66623 +146
- Misses 22470 22537 +67
Partials 1170 1170
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `83.63% <ø> (-0.06%)` | :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/17429?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/util.go](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS91dGlsLmdv) | `29.54% <ø> (ø)` | |
| [sdks/python/apache\_beam/utils/interactive\_utils.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdXRpbHMvaW50ZXJhY3RpdmVfdXRpbHMucHk=) | `90.24% <0.00%> (-4.88%)` | :arrow_down: |
| [...pache\_beam/runners/interactive/interactive\_beam.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9pbnRlcmFjdGl2ZV9iZWFtLnB5) | `77.81% <0.00%> (-4.00%)` | :arrow_down: |
| [...che\_beam/runners/interactive/interactive\_runner.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9pbnRlcmFjdGl2ZV9ydW5uZXIucHk=) | `90.84% <0.00%> (-2.12%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/source\_test\_utils.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vc291cmNlX3Rlc3RfdXRpbHMucHk=) | `88.01% <0.00%> (-1.39%)` | :arrow_down: |
| [...ache\_beam/runners/portability/expansion\_service.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9leHBhbnNpb25fc2VydmljZS5weQ==) | `91.83% <0.00%> (-1.35%)` | :arrow_down: |
| [...eam/runners/portability/fn\_api\_runner/execution.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9mbl9hcGlfcnVubmVyL2V4ZWN1dGlvbi5weQ==) | `92.25% <0.00%> (-0.81%)` | :arrow_down: |
| [sdks/python/apache\_beam/transforms/core.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9jb3JlLnB5) | `91.84% <0.00%> (-0.81%)` | :arrow_down: |
| [...runners/interactive/display/pcoll\_visualization.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9kaXNwbGF5L3Bjb2xsX3Zpc3VhbGl6YXRpb24ucHk=) | `85.85% <0.00%> (-0.51%)` | :arrow_down: |
| [sdks/python/apache\_beam/transforms/combiners.py](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9jb21iaW5lcnMucHk=) | `93.03% <0.00%> (-0.39%)` | :arrow_down: |
| ... and [10 more](https://codecov.io/gh/apache/beam/pull/17429/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/17429?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/17429?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [73c683f...3287c3c](https://codecov.io/gh/apache/beam/pull/17429?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863830842
##########
sdks/go/cmd/specialize/main.go:
##########
@@ -238,3 +246,130 @@ func upto(i int) []int {
}
return ret
}
+
+func add(i int, j int) int {
+ return i + j
+}
+
+func mult(i int, j int) int {
+ return i * j
+}
+
+func dict(values ...interface{}) map[string]interface{} {
+ dict := make(map[string]interface{}, len(values)/2)
+ if len(values)%2 != 0 {
+ panic("Invalid dictionary call")
+ }
+ for i := 0; i < len(values); i += 2 {
+ dict[values[i].(string)] = values[i+1]
+ }
+
+ return dict
+}
+
+func list(values ...string) []string {
+ return values
+}
+
+func genericTypingRepresentation(in int, out int, includeType bool) string {
+ seenElements := false
+ typing := ""
+ if in > 0 {
+ typing += fmt.Sprintf("[I%v", 0)
+ for i := 1; i < in; i++ {
+ typing += fmt.Sprintf(", I%v", i)
+ }
+ seenElements = true
+ }
+ if out > 0 {
+ i := 0
+ if !seenElements {
+ typing += fmt.Sprintf("[R%v", 0)
+ i++
+ }
+ for i < out {
+ typing += fmt.Sprintf(", R%v", i)
+ i++
+ }
+ seenElements = true
+ }
+
+ if seenElements {
+ if includeType {
+ typing += " any"
+ }
+ typing += "]"
+ }
+
+ return typing
+}
+
+func possibleBundleLifecycleParameterCombos(numInInterface interface{}, processElementInInterface interface{}) [][]string {
+ numIn := numInInterface.(int)
+ processElementIn := processElementInInterface.(int)
+ ordered_known_parameter_options := []string{"context.Context", "typex.PaneInfo", "[]typex.Window", "typex.EventTime", "typex.BundleFinalization"}
+ // Because of how Bundle lifecycle functions are invoked, all known parameters must preced unknown options and be in order.
Review Comment:
I think we're both wrong and its precede haha
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863988306
##########
sdks/go/pkg/beam/registration/example_registration_test.go:
##########
@@ -0,0 +1,51 @@
+// 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 registration_test
+
+import (
+ "context"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/registration"
+)
+
+type myDoFn struct {
+}
+
+func (fn *myDoFn) ProcessElement(word string, emit func(int)) int {
+ emit(len(word))
+ return len(word)
+}
+
+func (fn *myDoFn) StartBundle(_ context.Context, emit func(int)) {
+ emit(2)
+}
+
+func (fn *myDoFn) FinishBundle(_ context.Context, emit func(int)) error {
+ emit(2)
+ return nil
+}
+
+func (fn *myDoFn) Setup(_ context.Context) error {
+ return nil
+}
+
+func (fn *myDoFn) Teardown() error {
+ return nil
+}
+
+func ExampleDoFn2x1() {
+ registration.DoFn2x1[string, func(int), int](&myDoFn{})
Review Comment:
It may be good to add a comment here explaining why myDoFn needs to be called with DoFn2x1 rather than anything else.
##########
sdks/go/cmd/specialize/main_test.go:
##########
@@ -0,0 +1,146 @@
+// 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 main
+
+import (
+ "fmt"
+ "testing"
+)
+
+func TestGenericTypingRepresentation(t *testing.T) {
+ tests := []struct {
+ in int
+ out int
+ includeType bool
+ representation string
+ }{
+ {0, 0, true, ""},
+ {0, 0, false, ""},
+ {0, 1, true, "[R0 any]"},
+ {0, 1, false, "[R0]"},
+ {0, 3, true, "[R0, R1, R2 any]"},
+ {0, 3, false, "[R0, R1, R2]"},
+ {1, 0, true, "[I0 any]"},
+ {1, 0, false, "[I0]"},
+ {3, 0, true, "[I0, I1, I2 any]"},
+ {3, 0, false, "[I0, I1, I2]"},
+ {1, 1, true, "[I0, R0 any]"},
+ {1, 1, false, "[I0, R0]"},
+ {4, 1, true, "[I0, I1, I2, I3, R0 any]"},
+ {4, 1, false, "[I0, I1, I2, I3, R0]"},
+ {4, 2, true, "[I0, I1, I2, I3, R0, R1 any]"},
+ {4, 2, false, "[I0, I1, I2, I3, R0, R1]"},
+ {1, 2, true, "[I0, R0, R1 any]"},
+ {1, 2, false, "[I0, R0, R1]"},
+ }
+
+ for _, test := range tests {
+ if got, want := genericTypingRepresentation(test.in, test.out, test.includeType), test.representation; got != want {
+ t.Errorf("genericTypingRepresentation(%v, %v, %v) = %v, want: %v", test.in, test.out, test.includeType, got, want)
+ }
+ }
+}
+
+func TestPossibleBundleLifecycleParameterCombos(t *testing.T) {
+ tests := []struct {
+ numIn int
+ processElementIn int
+ representation [][]string
+ }{
+ {0, 0, [][]string{[]string{}}},
+ {0, 1, [][]string{[]string{}}},
+ {1, 0, [][]string{[]string{"context.Context"}, []string{"typex.PaneInfo"}, []string{"[]typex.Window"}, []string{"typex.EventTime"}, []string{"typex.BundleFinalization"}}},
+ {1, 1, [][]string{[]string{"I0"}, []string{"context.Context"}, []string{"typex.PaneInfo"}, []string{"[]typex.Window"}, []string{"typex.EventTime"}, []string{"typex.BundleFinalization"}}},
+ {2, 1, [][]string{[]string{"context.Context", "I0"}, []string{"context.Context", "typex.PaneInfo"}, []string{"context.Context", "[]typex.Window"}, []string{"context.Context", "typex.EventTime"}, []string{"context.Context", "typex.BundleFinalization"},
+ []string{"typex.PaneInfo", "I0"}, []string{"typex.PaneInfo", "[]typex.Window"}, []string{"typex.PaneInfo", "typex.EventTime"}, []string{"typex.PaneInfo", "typex.BundleFinalization"},
+ []string{"[]typex.Window", "I0"}, []string{"[]typex.Window", "typex.EventTime"}, []string{"[]typex.Window", "typex.BundleFinalization"},
+ []string{"typex.EventTime", "I0"}, []string{"typex.EventTime", "typex.BundleFinalization"},
+ []string{"typex.BundleFinalization", "I0"}}},
+ {2, 2, [][]string{[]string{"context.Context", "I1"}, []string{"context.Context", "typex.PaneInfo"}, []string{"context.Context", "[]typex.Window"}, []string{"context.Context", "typex.EventTime"}, []string{"context.Context", "typex.BundleFinalization"},
+ []string{"typex.PaneInfo", "I1"}, []string{"typex.PaneInfo", "[]typex.Window"}, []string{"typex.PaneInfo", "typex.EventTime"}, []string{"typex.PaneInfo", "typex.BundleFinalization"},
+ []string{"[]typex.Window", "I1"}, []string{"[]typex.Window", "typex.EventTime"}, []string{"[]typex.Window", "typex.BundleFinalization"},
+ []string{"typex.EventTime", "I1"}, []string{"typex.EventTime", "typex.BundleFinalization"},
+ []string{"typex.BundleFinalization", "I1"},
+ []string{"I0", "I1"}}},
+ {2, 3, [][]string{[]string{"context.Context", "I2"}, []string{"context.Context", "typex.PaneInfo"}, []string{"context.Context", "[]typex.Window"}, []string{"context.Context", "typex.EventTime"}, []string{"context.Context", "typex.BundleFinalization"},
+ []string{"typex.PaneInfo", "I2"}, []string{"typex.PaneInfo", "[]typex.Window"}, []string{"typex.PaneInfo", "typex.EventTime"}, []string{"typex.PaneInfo", "typex.BundleFinalization"},
+ []string{"[]typex.Window", "I2"}, []string{"[]typex.Window", "typex.EventTime"}, []string{"[]typex.Window", "typex.BundleFinalization"},
+ []string{"typex.EventTime", "I2"}, []string{"typex.EventTime", "typex.BundleFinalization"},
+ []string{"typex.BundleFinalization", "I2"},
+ []string{"I1", "I2"}}},
+ }
+
+ for _, test := range tests {
+ got, want := possibleBundleLifecycleParameterCombos(test.numIn, test.processElementIn), test.representation
+ assertRepresentationEquals(t, got, want, fmt.Sprintf("possibleBundleLifecycleParameterCombos(%v, %v)", test.numIn, test.processElementIn))
+ }
+}
+
+func TestMakeStructRegisterEntry(t *testing.T) {
+ tests := []struct {
+ structName string
+ functionName string
+ ins []string
+ outs []string
+ entry string
+ }{
+ {"StartBundle", "startBundle", []string{}, []string{}, "reflectx.MakeFunc(func() { fn.(StartBundle0x0).startBundle() })"},
+ {"StartBundle", "startBundle", []string{"context.Context"}, []string{}, "reflectx.MakeFunc(func(a0 context.Context) { fn.(StartBundle1x0[context.Context]).startBundle(a0) })"},
+ {"StartBundle", "startBundle", []string{"context.Context", "I0"}, []string{}, "reflectx.MakeFunc(func(a0 context.Context, a1 I0) { fn.(StartBundle2x0[context.Context, I0]).startBundle(a0, a1) })"},
+ {"FinishBundle", "finishBundle", []string{}, []string{"[]typex.Window"}, "reflectx.MakeFunc(func() ([]typex.Window){ return fn.(FinishBundle0x1[[]typex.Window]).finishBundle() })"},
+ {"FinishBundle", "finishBundle", []string{}, []string{"[]typex.Window", "R0"}, "reflectx.MakeFunc(func() ([]typex.Window, R0){ return fn.(FinishBundle0x2[[]typex.Window, R0]).finishBundle() })"},
+ {"FinishBundle", "finishBundle", []string{"I0"}, []string{"[]typex.Window", "R0"}, "reflectx.MakeFunc(func(a0 I0) ([]typex.Window, R0){ return fn.(FinishBundle1x2[I0, []typex.Window, R0]).finishBundle(a0) })"},
+ {"FinishBundle", "finishBundle", []string{"context.Context", "I0"}, []string{"[]typex.Window", "R0"}, "reflectx.MakeFunc(func(a0 context.Context, a1 I0) ([]typex.Window, R0){ return fn.(FinishBundle2x2[context.Context, I0, []typex.Window, R0]).finishBundle(a0, a1) })"},
+ }
+
+ for _, test := range tests {
+ if got, want := makeStructRegisterEntry(test.structName, test.functionName, test.ins, test.outs), test.entry; got != want {
+ t.Errorf("makeStructRegisterEntry(\"%v\", \"%v\", %v, %v) = %v, want: %v", test.structName, test.functionName, test.ins, test.outs, got, want)
+ }
+ }
+}
+
+func assertRepresentationEquals(t *testing.T, got [][]string, want [][]string, functionInvocation string) {
Review Comment:
Add a call to t.Helper(), so the failure is ascribed to the call site and not in this function.
Consider instead not having the assert function and moving the body into the test, since it looks like you ended up with just the one call site?
I'll note that you may also wish to look at the cmp package, which may clean this up a little bit WRT the comparison. https://pkg.go.dev/github.com/google/go-cmp/cmp
##########
sdks/go/pkg/beam/registration/registration.go:
##########
@@ -0,0 +1,6990 @@
+// File generated by specialize. Do not edit.
+
+// 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.
+
+// Code generated from registration.tmpl. DO NOT EDIT.
+
+package registration
+
+import (
+ "context"
+ "reflect"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
+)
+
+func registerStartBundle0x0FuncAndMakeStructWrapper() func(interface{}) reflectx.Func {
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), func(fn interface{}) reflectx.Func {
+ return &caller0x0{fn: fn.(func())}
+ })
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.(startBundle0x0).StartBundle()
+ })
+ }
+}
+
+func registerStartBundle0x1FuncAndMakeStructWrapper[R0 any]() func(interface{}) reflectx.Func {
Review Comment:
This structure makes it clear that the factory structure doesn't even require any parameters passed in, just the types. That's awesome!
##########
sdks/go/pkg/beam/registration/doc.go:
##########
@@ -0,0 +1,31 @@
+// 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.
+
+//go:generate go install github.com/apache/beam/sdks/v2/go/cmd/specialize
+//go:generate specialize --package=registration --input=registration.tmpl --x=data,universals --imports=typex
+//go:generate go fmt
+
+/*
+Package registration contains functions for registering and optimizing your DoFn.
+
+This package contains generic registration/optimization function for each possible combination of input and output arities in a DoFn's ProcessElement function.
+For example, given a DoFn with a ProcessElement function that takes 4 inputs and returns 3 outputs, you can call
+register.DoFn4x3[input1 type, input2 type, input3 type, input4 type, output1 type, output2 type, output3 type](&doFn{}) during pipeline construction. This will
+register your DoFn and perform some reflection based optimization to significantly speed up execution at runtime.
+
+See DoFn2x1 for a full example.
+*/
+
Review Comment:
rm this blank line. The line break prevent associating this comment with the package statement, applies to all Go Doc comments. Floating comments live only in the code, which is why it's important to have a break after the license, but before the comment so the licence doesn't go into the generated doc.
##########
sdks/go/pkg/beam/registration/doc.go:
##########
@@ -0,0 +1,31 @@
+// 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.
+
+//go:generate go install github.com/apache/beam/sdks/v2/go/cmd/specialize
+//go:generate specialize --package=registration --input=registration.tmpl --x=data,universals --imports=typex
+//go:generate go fmt
+
+/*
+Package registration contains functions for registering and optimizing your DoFn.
+
+This package contains generic registration/optimization function for each possible combination of input and output arities in a DoFn's ProcessElement function.
+For example, given a DoFn with a ProcessElement function that takes 4 inputs and returns 3 outputs, you can call
+register.DoFn4x3[input1 type, input2 type, input3 type, input4 type, output1 type, output2 type, output3 type](&doFn{}) during pipeline construction. This will
+register your DoFn and perform some reflection based optimization to significantly speed up execution at runtime.
Review Comment:
Technically the "reflection based" optimization we're doing is "avoiding using reflection entirely", so it feels off to call it that.
Maybe `...produces optimized callers for your DoFn to significantly speed up execution at runtime` instead?
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,263 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $funcIn := upto $startFinishBundleMaxIn}}{{range $funcOut := upto 2}}
+func register{{$upperName}}{{$funcIn}}x{{$funcOut}}FuncAndMakeStructWrapper{{(genericTypingRepresentation $funcIn $funcOut true)}}() func(interface{}) reflectx.Func {
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{range $in := upto $funcIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $funcOut}} R0{{end}})(nil)).Elem(), func(fn interface{}) reflectx.Func {
+ return &caller{{$funcIn}}x{{$funcOut}}{{(genericTypingRepresentation $funcIn $funcOut false)}}{fn: fn.(func({{range $in := upto $funcIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $funcOut}} R0{{end}})}
+ })
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func({{range $in := upto $funcIn}}{{if $in}}, {{end}}a{{$in}} I{{$in}}{{end}}){{if $funcOut}} R0{{end}} {
+ {{if $funcOut}}return {{end}}fn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{(genericTypingRepresentation $funcIn $funcOut false)}}).{{$upperName}}({{range $in := upto $funcIn}}{{if $in}}, {{end}}a{{$in}}{{end}})
+ })
+ }
+}
+{{end}}{{end}}
+{{$numParams := $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
+ {{$lowerName}}In := -1
+ {{$lowerName}}Out := -1
+ {{$lowerName}}Method := reflect.ValueOf(doFn).MethodByName("{{$upperName}}")
+ if {{$lowerName}}Method.IsValid() {
+ {{$lowerName}}In = {{$lowerName}}Method.Type().NumIn()
+ {{$lowerName}}Out = {{$lowerName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto $startFinishBundleMaxIn}}
+ case {{$lowerName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $numParams)}}{{if $possibleCombos}}
+ case {{$lowerName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ return register{{$upperName}}{{$funcIn}}x{{$funcOut}}FuncAndMakeStructWrapper{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}()
+ } {{end}}{{end}}{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+ }
+ return nil
Review Comment:
If I followed this correctly in the generated file, it looks like it might be possible for a dense innermost if/else ladder to return out, and land at this `return nil`. Is this intentional?
If so, we should just have a comment just above this saying when the nil return is expected to happen, especially since the switch defaults are to panic.
If not, either add a terminal else clause to the inner if/else ladder, or turn it to a switch so it can also have a terminal default.
##########
sdks/go/pkg/beam/registration/registration.go:
##########
@@ -0,0 +1,6990 @@
+// File generated by specialize. Do not edit.
Review Comment:
I'm just excited that the new file/diff is now small enough to show up in the diff UI. :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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1117181488
Run Java_PVR_Flink_Batch PreCommit
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1105640985
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
lostluck commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r859383238
##########
sdks/go/cmd/specialize/main.go:
##########
@@ -238,3 +246,133 @@ func upto(i int) []int {
}
return ret
}
+
+func add(i int, j int) int {
+ return i + j
+}
+
+func mult(i int, j int) int {
+ return i * j
+}
+
+func dict(values ...interface{}) map[string]interface{} {
+ dict := make(map[string]interface{}, len(values)/2)
+ if len(values)%2 != 0 {
+ panic("Invalid dictionary call")
+ }
+ for i := 0; i < len(values); i += 2 {
+ dict[values[i].(string)] = values[i+1]
+ }
+
+ return dict
+}
+
+func list(values ...string) []string {
+ return values
+}
+
+func genericTypingRepresentation(in int, out int, includeType bool) string {
+ seenElements := false
+ typing := ""
+ if in > 0 {
+ typing += fmt.Sprintf("[I%v", 0)
+ for i := 1; i < in; i++ {
+ typing += fmt.Sprintf(", I%v", i)
+ }
+ seenElements = true
+ }
+ if out > 0 {
+ i := 0
+ if !seenElements {
+ typing += fmt.Sprintf("[R%v", 0)
+ i++
+ }
+ for i < out {
+ typing += fmt.Sprintf(", R%v", i)
+ i++
+ }
+ seenElements = true
+ }
+
+ if seenElements {
+ if includeType {
+ typing += " any"
+ }
+ typing += "]"
+ }
+
+ return typing
+}
+
+func possibleBundleLifecycleParameterCombos(numInInterface interface{}, processElementInInterface interface{}) [][]string {
+ numIn := numInInterface.(int)
+ processElementIn := processElementInInterface.(int)
+ ordered_known_parameter_options := []string{"context.Context", "typex.PaneInfo", "[]typex.Window", "typex.EventTime", "typex.BundleFinalization"}
+ // Because of how Bundle lifecycle functions are invoked, all known parameters must preced unknown options and be in order.
+ // Once we hit an unknown options, all remaining unknown options must be included since all iters/emitters must be included
+ // Therefore, we can generate a powerset of the known options and fill out any remaining parameters with an ordered set of remaining unknown options
+ pSetSize := int(math.Pow(2, float64(len(ordered_known_parameter_options))))
+ combos := make([][]string, 0, pSetSize)
+
+ var index int
+ for index < pSetSize {
Review Comment:
There doesn't seem to be any reason not to use a standard `for index := 0; index < pSetSize; index++ {` construct here.
(Don't forget to remove the `index++` from the bottom of the loop.)
##########
sdks/go/pkg/beam/registration.tmpl:
##########
@@ -0,0 +1,292 @@
+{{define "StructWrappersAndFuncs_StartFinishBundle"}}
Review Comment:
I strongly recommend we have the registration methods in *different* package than the main user surface beam package. It would make the package doc much harder to use.
If the package is called `register` then the methods become `register.DoFnNxMxO` etc. which read pretty cleanly. This also avoids forcing generics into the main beam package.
##########
sdks/go/pkg/beam/registration.tmpl:
##########
@@ -0,0 +1,292 @@
+{{define "StructWrappersAndFuncs_StartFinishBundle"}}
+{{$funcName := "unknown"}}{{$structName := "unknown"}}{{if (eq .func "startBundle")}}{{$funcName = "startBundle"}}{{$structName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$funcName = "finishBundle"}}{{$structName = "FinishBundle"}}{{end}}
+ {{$funcName}}In := -1
+ {{$funcName}}Out := -1
+ var {{$funcName}}Wrapper func(fn interface{}) reflectx.Func
+ {{$funcName}}Method := reflect.ValueOf(doFn).MethodByName("{{$structName}}")
+ if {{$funcName}}Method.IsValid() {
+ {{$funcName}}In = {{$funcName}}Method.Type().NumIn()
+ {{$funcName}}Out = {{$funcName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto 8}}
+ case {{$funcName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $.processElementIn)}}{{if $possibleCombos}}
+ case {{$funcName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$funcName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ {{$funcName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})
+ return &caller{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})(nil)).Elem(), {{$funcName}}Caller)
+ {{$funcRegister := (makeStructRegisterEntry $funcName $structName $funcCombo (list))}}{{if $funcOut}}{{$funcRegister = (makeStructRegisterEntry $funcName $structName $funcCombo (list "error"))}}{{end}}
+ {{$funcName}}Wrapper = func(fn interface{}) reflectx.Func {
+ return {{$funcRegister}}
+ }
+ } {{end}}{{end}}{{end}}
+ default:
+ panic("Invalid signature for {{$structName}}")
+ }
+{{end}}
+ default:
+ panic("Invalid signature for {{$structName}}")
+ }
+ }
+{{end}}
+// 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.
+
+// Code generated from registration.tmpl. DO NOT EDIT.
+
+package beam
+
+import (
+ "context"
+ "reflect"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
+)
+{{$processElementMaxOut := 5}}{{$processElementMaxIn := 9}}{{$startFinishBundleOutRange := 2}}{{$startFinishBundleInRange := 8}}{{range $processElementOut := upto $processElementMaxOut}}{{range $processElementIn := upto $processElementMaxIn}}
+type doFn{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut true)}} interface {
+ ProcessElement({{range $in := upto $processElementIn}}{{if $in}}, {{end}}i{{$in}} I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}}
+}
+
+type caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut true)}} struct {
+ fn func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}}
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Name() string {
+ return reflectx.FunctionName(c.fn)
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Type() reflect.Type {
+ return reflect.TypeOf(c.fn)
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Call(args []interface{}) []interface{} {
+ {{if $processElementOut}}{{range $out := upto $processElementOut}}{{if $out}}, {{end}}out{{$out}}{{end}} := {{end}}c.fn({{range $in := upto $processElementIn}}{{if $in}}, {{end}}args[{{$in}}].(I{{$in}}){{end}})
+ return []interface{}{ {{if $processElementOut}}{{range $out := upto $processElementOut}}{{if $out}}, {{end}}out{{$out}}{{end}}{{end}} }
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Call{{$processElementIn}}x{{$processElementOut}}({{range $in := upto $processElementIn}}{{if $in}}, {{end}}arg{{$in}} I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}interface{}{{end}}){{end}} {
+ {{if $processElementOut}}return {{end}}c.fn({{range $in := upto $processElementIn}}{{if $in}}, {{end}}arg{{$in}}{{end}})
+}
+
+func registerDoFn{{$processElementIn}}x{{$processElementOut}}StructWrappersAndFuncs{{(genericTypingRepresentation $processElementIn $processElementOut true)}}(doFn doFn{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) {
+ processElementCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}})
+ return &caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}})(nil)).Elem(), processElementCaller)
+ processElementWrapper := func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}a{{$in}} I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}} {
+ {{if $processElementOut}}return {{end}}fn.(doFn{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}).ProcessElement({{range $in := upto $processElementIn}}{{if $in}}, {{end}}a{{$in}}{{end}})
+ })
+ }{{template "StructWrappersAndFuncs_StartFinishBundle" (dict "processElementIn" $processElementIn "processElementOut" $processElementOut "func" "startBundle")}}{{template "StructWrappersAndFuncs_StartFinishBundle" (dict "processElementIn" $processElementIn "processElementOut" $processElementOut "func" "finishBundle")}}
+ var setupWrapper func(fn interface{}) reflectx.Func
+ if _, ok := doFn.(setup0x0); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func())
+ return &caller0x0{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.(setup0x0).Setup()
+ })
+ }
+ } else if _, ok := doFn.(setup1x0); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context))
+ return &caller1x0[context.Context]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context))(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) {
+ fn.(setup1x0).Setup(a0)
+ })
+ }
+ } else if _, ok := doFn.(setup0x1); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func() error)
+ return &caller0x1[error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func() error)(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() error {
+ return fn.(setup0x1).Setup()
+ })
+ }
+ } else if _, ok := doFn.(setup1x1); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context) error)
+ return &caller1x1[context.Context, error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context) error)(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) error {
+ return fn.(setup1x1).Setup(a0)
+ })
+ }
+ }
+ var teardownWrapper func(fn interface{}) reflectx.Func
+ if _, ok := doFn.(teardown0x0); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func())
+ return &caller0x0{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.(teardown0x0).Teardown()
+ })
+ }
+ } else if _, ok := doFn.(teardown1x0); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context))
+ return &caller1x0[context.Context]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context))(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) {
+ fn.(teardown1x0).Teardown(a0)
+ })
+ }
+ } else if _, ok := doFn.(teardown0x1); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func() error)
+ return &caller0x1[error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func() error)(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() error {
+ return fn.(teardown0x1).Teardown()
+ })
+ }
+ } else if _, ok := doFn.(teardown1x1); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context) error)
+ return &caller1x1[context.Context, error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context) error)(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) error {
+ return fn.(teardown1x1).Teardown(a0)
+ })
+ }
+ }
+ wrapperFn := func(fn interface{}) map[string]reflectx.Func {
+ m := map[string]reflectx.Func{}
+ if processElementWrapper != nil {
+ m["ProcessElement"] = processElementWrapper(fn)
+ }
+ if startBundleWrapper != nil {
+ m["StartBundle"] = startBundleWrapper(fn)
+ }
+ if finishBundleWrapper != nil {
+ m["FinishBundle"] = finishBundleWrapper(fn)
+ }
+ if setupWrapper != nil {
+ m["Setup"] = setupWrapper(fn)
+ }
+ if teardownWrapper != nil {
+ m["Teardown"] = teardownWrapper(fn)
+ }
+
+ return m
+ }
Review Comment:
This is probably something you had in mind for helping reduce the size of the generated code since a quick skim reads pretty static...
Do recall that if a helper doesn't need arity assistance, it may be able to be put into a non-generated file the same package, and tested normally.
--
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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r860058545
##########
sdks/go/pkg/beam/registration.tmpl:
##########
@@ -0,0 +1,292 @@
+{{define "StructWrappersAndFuncs_StartFinishBundle"}}
+{{$funcName := "unknown"}}{{$structName := "unknown"}}{{if (eq .func "startBundle")}}{{$funcName = "startBundle"}}{{$structName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$funcName = "finishBundle"}}{{$structName = "FinishBundle"}}{{end}}
+ {{$funcName}}In := -1
+ {{$funcName}}Out := -1
+ var {{$funcName}}Wrapper func(fn interface{}) reflectx.Func
+ {{$funcName}}Method := reflect.ValueOf(doFn).MethodByName("{{$structName}}")
+ if {{$funcName}}Method.IsValid() {
+ {{$funcName}}In = {{$funcName}}Method.Type().NumIn()
+ {{$funcName}}Out = {{$funcName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto 8}}
+ case {{$funcName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $.processElementIn)}}{{if $possibleCombos}}
+ case {{$funcName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$funcName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ {{$funcName}}Caller := func(fn interface{}) reflectx.Func {
+ f := fn.(func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})
+ return &caller{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})(nil)).Elem(), {{$funcName}}Caller)
+ {{$funcRegister := (makeStructRegisterEntry $funcName $structName $funcCombo (list))}}{{if $funcOut}}{{$funcRegister = (makeStructRegisterEntry $funcName $structName $funcCombo (list "error"))}}{{end}}
+ {{$funcName}}Wrapper = func(fn interface{}) reflectx.Func {
+ return {{$funcRegister}}
+ }
+ } {{end}}{{end}}{{end}}
+ default:
+ panic("Invalid signature for {{$structName}}")
+ }
+{{end}}
+ default:
+ panic("Invalid signature for {{$structName}}")
+ }
+ }
+{{end}}
+// 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.
+
+// Code generated from registration.tmpl. DO NOT EDIT.
+
+package beam
+
+import (
+ "context"
+ "reflect"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/graphx/schema"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/reflectx"
+)
+{{$processElementMaxOut := 5}}{{$processElementMaxIn := 9}}{{$startFinishBundleOutRange := 2}}{{$startFinishBundleInRange := 8}}{{range $processElementOut := upto $processElementMaxOut}}{{range $processElementIn := upto $processElementMaxIn}}
+type doFn{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut true)}} interface {
+ ProcessElement({{range $in := upto $processElementIn}}{{if $in}}, {{end}}i{{$in}} I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}}
+}
+
+type caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut true)}} struct {
+ fn func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}}
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Name() string {
+ return reflectx.FunctionName(c.fn)
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Type() reflect.Type {
+ return reflect.TypeOf(c.fn)
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Call(args []interface{}) []interface{} {
+ {{if $processElementOut}}{{range $out := upto $processElementOut}}{{if $out}}, {{end}}out{{$out}}{{end}} := {{end}}c.fn({{range $in := upto $processElementIn}}{{if $in}}, {{end}}args[{{$in}}].(I{{$in}}){{end}})
+ return []interface{}{ {{if $processElementOut}}{{range $out := upto $processElementOut}}{{if $out}}, {{end}}out{{$out}}{{end}}{{end}} }
+}
+
+func (c *caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) Call{{$processElementIn}}x{{$processElementOut}}({{range $in := upto $processElementIn}}{{if $in}}, {{end}}arg{{$in}} I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}interface{}{{end}}){{end}} {
+ {{if $processElementOut}}return {{end}}c.fn({{range $in := upto $processElementIn}}{{if $in}}, {{end}}arg{{$in}}{{end}})
+}
+
+func registerDoFn{{$processElementIn}}x{{$processElementOut}}StructWrappersAndFuncs{{(genericTypingRepresentation $processElementIn $processElementOut true)}}(doFn doFn{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}) {
+ processElementCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}})
+ return &caller{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}})(nil)).Elem(), processElementCaller)
+ processElementWrapper := func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func({{range $in := upto $processElementIn}}{{if $in}}, {{end}}a{{$in}} I{{$in}}{{end}}){{if $processElementOut}} ({{range $out := upto $processElementOut}}{{if $out}}, {{end}}R{{$out}}{{end}}){{end}} {
+ {{if $processElementOut}}return {{end}}fn.(doFn{{$processElementIn}}x{{$processElementOut}}{{(genericTypingRepresentation $processElementIn $processElementOut false)}}).ProcessElement({{range $in := upto $processElementIn}}{{if $in}}, {{end}}a{{$in}}{{end}})
+ })
+ }{{template "StructWrappersAndFuncs_StartFinishBundle" (dict "processElementIn" $processElementIn "processElementOut" $processElementOut "func" "startBundle")}}{{template "StructWrappersAndFuncs_StartFinishBundle" (dict "processElementIn" $processElementIn "processElementOut" $processElementOut "func" "finishBundle")}}
+ var setupWrapper func(fn interface{}) reflectx.Func
+ if _, ok := doFn.(setup0x0); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func())
+ return &caller0x0{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.(setup0x0).Setup()
+ })
+ }
+ } else if _, ok := doFn.(setup1x0); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context))
+ return &caller1x0[context.Context]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context))(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) {
+ fn.(setup1x0).Setup(a0)
+ })
+ }
+ } else if _, ok := doFn.(setup0x1); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func() error)
+ return &caller0x1[error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func() error)(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() error {
+ return fn.(setup0x1).Setup()
+ })
+ }
+ } else if _, ok := doFn.(setup1x1); ok {
+ setupCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context) error)
+ return &caller1x1[context.Context, error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context) error)(nil)).Elem(), setupCaller)
+
+ setupWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) error {
+ return fn.(setup1x1).Setup(a0)
+ })
+ }
+ }
+ var teardownWrapper func(fn interface{}) reflectx.Func
+ if _, ok := doFn.(teardown0x0); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func())
+ return &caller0x0{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func())(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() {
+ fn.(teardown0x0).Teardown()
+ })
+ }
+ } else if _, ok := doFn.(teardown1x0); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context))
+ return &caller1x0[context.Context]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context))(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) {
+ fn.(teardown1x0).Teardown(a0)
+ })
+ }
+ } else if _, ok := doFn.(teardown0x1); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func() error)
+ return &caller0x1[error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func() error)(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func() error {
+ return fn.(teardown0x1).Teardown()
+ })
+ }
+ } else if _, ok := doFn.(teardown1x1); ok {
+ teardownCaller := func(fn interface{}) reflectx.Func {
+ f := fn.(func(context.Context) error)
+ return &caller1x1[context.Context, error]{fn: f}
+ }
+ reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context) error)(nil)).Elem(), teardownCaller)
+
+ teardownWrapper = func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func(a0 context.Context) error {
+ return fn.(teardown1x1).Teardown(a0)
+ })
+ }
+ }
+ wrapperFn := func(fn interface{}) map[string]reflectx.Func {
+ m := map[string]reflectx.Func{}
+ if processElementWrapper != nil {
+ m["ProcessElement"] = processElementWrapper(fn)
+ }
+ if startBundleWrapper != nil {
+ m["StartBundle"] = startBundleWrapper(fn)
+ }
+ if finishBundleWrapper != nil {
+ m["FinishBundle"] = finishBundleWrapper(fn)
+ }
+ if setupWrapper != nil {
+ m["Setup"] = setupWrapper(fn)
+ }
+ if teardownWrapper != nil {
+ m["Teardown"] = teardownWrapper(fn)
+ }
+
+ return m
+ }
Review Comment:
This was not actually one the big one that I was considering, though I did change it as well
> Do recall that if a helper doesn't need arity assistance, it may be able to be put into a non-generated file the same package, and tested normally.
That brings up an interesting style question - I had considered doing this initially, but decided to keep it in the template file because logically it belongs with the other functions in registration.go. Stylistically, is it more typical to move functions like this out of the template file or to keep them with the functions they logically go with? Doesn't really matter for this case at this point, but I'm curious about the principle
##########
sdks/go/cmd/specialize/main.go:
##########
@@ -238,3 +246,133 @@ func upto(i int) []int {
}
return ret
}
+
+func add(i int, j int) int {
+ return i + j
+}
+
+func mult(i int, j int) int {
+ return i * j
+}
+
+func dict(values ...interface{}) map[string]interface{} {
+ dict := make(map[string]interface{}, len(values)/2)
+ if len(values)%2 != 0 {
+ panic("Invalid dictionary call")
+ }
+ for i := 0; i < len(values); i += 2 {
+ dict[values[i].(string)] = values[i+1]
+ }
+
+ return dict
+}
+
+func list(values ...string) []string {
+ return values
+}
+
+func genericTypingRepresentation(in int, out int, includeType bool) string {
+ seenElements := false
+ typing := ""
+ if in > 0 {
+ typing += fmt.Sprintf("[I%v", 0)
+ for i := 1; i < in; i++ {
+ typing += fmt.Sprintf(", I%v", i)
+ }
+ seenElements = true
+ }
+ if out > 0 {
+ i := 0
+ if !seenElements {
+ typing += fmt.Sprintf("[R%v", 0)
+ i++
+ }
+ for i < out {
+ typing += fmt.Sprintf(", R%v", i)
+ i++
+ }
+ seenElements = true
+ }
+
+ if seenElements {
+ if includeType {
+ typing += " any"
+ }
+ typing += "]"
+ }
+
+ return typing
+}
+
+func possibleBundleLifecycleParameterCombos(numInInterface interface{}, processElementInInterface interface{}) [][]string {
+ numIn := numInInterface.(int)
+ processElementIn := processElementInInterface.(int)
+ ordered_known_parameter_options := []string{"context.Context", "typex.PaneInfo", "[]typex.Window", "typex.EventTime", "typex.BundleFinalization"}
+ // Because of how Bundle lifecycle functions are invoked, all known parameters must preced unknown options and be in order.
+ // Once we hit an unknown options, all remaining unknown options must be included since all iters/emitters must be included
+ // Therefore, we can generate a powerset of the known options and fill out any remaining parameters with an ordered set of remaining unknown options
+ pSetSize := int(math.Pow(2, float64(len(ordered_known_parameter_options))))
+ combos := make([][]string, 0, pSetSize)
+
+ var index int
+ for index < pSetSize {
Review Comment:
There's not - this was a consequence of an evolving approach and I missed this bit of cleanup - thanks
##########
sdks/go/pkg/beam/registration.tmpl:
##########
@@ -0,0 +1,292 @@
+{{define "StructWrappersAndFuncs_StartFinishBundle"}}
Review Comment:
That's a fair concern - I primarily chose the main package because it is where our existing registration functions live, I don't think that is worth the cost to the docs 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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863310735
##########
.github/workflows/go_tests.yml:
##########
@@ -39,7 +39,7 @@ jobs:
fetch-depth: 2
- uses: actions/setup-go@v2
with:
- go-version: '1.16'
+ go-version: '1.18'
Review Comment:
Not really - we could stick it in a template, but I don't really think its worth it for cases like this, otherwise you end up with a proliferation of templates which can be tough to follow
--
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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863934884
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
Review Comment:
Its now actually viewable in GitHub!
--
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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r863934043
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
Review Comment:
Yeah, that helps a lot. Technically, this does introduce a little more comparison overhead (so the registration call will be slower), but not nearly as much as I originally thought (read: not a significant amount), and the reduced sdk size is worth it. The extra comparisons were the reason I didn't initially structure it this way.
--
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 #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1105763684
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1111272386
Its down to 60,000 lines which is still a lot, but its better at least
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1110326059
> LOL: "-45%" test coverage.
>
> +293,588 lines. Hmmm. When I look at this properly, I'll see if we can't figure out other helper functions to bring that down. I think that's larger than the rest of the SDK...
I think the right approach is to exclude it from coverage. I tried to do that in this PR, but I'm guessing it doesn't factor in the current changeset when calculating it.
With that said, shrinking this is a good goal by itself
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1111319941
@lostluck this should be ready for another look at this point!
--
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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r864075000
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,252 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $numParams := upto $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{$numParams}}{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
+ {{$lowerName}}In := -1
+ {{$lowerName}}Out := -1
+ {{$lowerName}}Method := reflect.ValueOf(doFn).MethodByName("{{$upperName}}")
+ if {{$lowerName}}Method.IsValid() {
+ {{$lowerName}}In = {{$lowerName}}Method.Type().NumIn()
+ {{$lowerName}}Out = {{$lowerName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto $startFinishBundleMaxIn}}
+ case {{$lowerName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $numParams)}}{{if $possibleCombos}}
+ case {{$lowerName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})(nil)).Elem(), func(fn interface{}) reflectx.Func { return &caller{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}{fn: fn.(func({{(join $funcCombo ", ")}}){{if $funcOut}} error{{end}})} })
+ {{$funcRegister := (makeStructRegisterEntry $lowerName $upperName $funcCombo (list))}}{{if $funcOut}}{{$funcRegister = (makeStructRegisterEntry $lowerName $upperName $funcCombo (list "error"))}}{{end}}
+ return func(fn interface{}) reflectx.Func { return {{$funcRegister}} }
Review Comment:
This did help significantly, even after the previous optimization - 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] damccorm commented on a diff in pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on code in PR #17429:
URL: https://github.com/apache/beam/pull/17429#discussion_r864441783
##########
sdks/go/pkg/beam/registration/registration.tmpl:
##########
@@ -0,0 +1,263 @@
+{{define "BuildWrapper_StartFinishBundle"}}
+{{$lowerName := "unknown"}}{{$upperName := "unknown"}}{{if (eq .func "startBundle")}}{{$lowerName = "startBundle"}}{{$upperName = "StartBundle"}}{{end}}{{if (eq .func "finishBundle")}}{{$lowerName = "finishBundle"}}{{$upperName = "FinishBundle"}}{{end}}{{$startFinishBundleMaxIn := .startFinishBundleMaxIn}}{{$processElementMaxIn := .processElementMaxIn}}
+{{range $funcIn := upto $startFinishBundleMaxIn}}{{range $funcOut := upto 2}}
+func register{{$upperName}}{{$funcIn}}x{{$funcOut}}FuncAndMakeStructWrapper{{(genericTypingRepresentation $funcIn $funcOut true)}}() func(interface{}) reflectx.Func {
+ reflectx.RegisterFunc(reflect.TypeOf((*func({{range $in := upto $funcIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $funcOut}} R0{{end}})(nil)).Elem(), func(fn interface{}) reflectx.Func {
+ return &caller{{$funcIn}}x{{$funcOut}}{{(genericTypingRepresentation $funcIn $funcOut false)}}{fn: fn.(func({{range $in := upto $funcIn}}{{if $in}}, {{end}}I{{$in}}{{end}}){{if $funcOut}} R0{{end}})}
+ })
+
+ return func(fn interface{}) reflectx.Func {
+ return reflectx.MakeFunc(func({{range $in := upto $funcIn}}{{if $in}}, {{end}}a{{$in}} I{{$in}}{{end}}){{if $funcOut}} R0{{end}} {
+ {{if $funcOut}}return {{end}}fn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{(genericTypingRepresentation $funcIn $funcOut false)}}).{{$upperName}}({{range $in := upto $funcIn}}{{if $in}}, {{end}}a{{$in}}{{end}})
+ })
+ }
+}
+{{end}}{{end}}
+{{$numParams := $processElementMaxIn}}
+func build{{$upperName}}Wrapper{{(genericTypingRepresentation $numParams 0 true)}}(doFn interface{}) func(interface{}) reflectx.Func {
+ {{$lowerName}}In := -1
+ {{$lowerName}}Out := -1
+ {{$lowerName}}Method := reflect.ValueOf(doFn).MethodByName("{{$upperName}}")
+ if {{$lowerName}}Method.IsValid() {
+ {{$lowerName}}In = {{$lowerName}}Method.Type().NumIn()
+ {{$lowerName}}Out = {{$lowerName}}Method.Type().NumOut()
+ switch {
+{{range $funcIn := upto $startFinishBundleMaxIn}}
+ case {{$lowerName}}In == {{$funcIn}}:
+ switch { {{range $funcOut := upto 2}}{{$possibleCombos := (possibleBundleLifecycleParameterCombos $funcIn $numParams)}}{{if $possibleCombos}}
+ case {{$lowerName}}Out == {{$funcOut}}:
+{{$first := true}}{{range $funcCombo := $possibleCombos}}{{if $first}}{{$first = false}} {{else}} else {{end}}if _, ok := doFn.({{$lowerName}}{{$funcIn}}x{{$funcOut}}{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}); ok {
+ return register{{$upperName}}{{$funcIn}}x{{$funcOut}}FuncAndMakeStructWrapper{{if (or $funcIn $funcOut)}}[{{(join $funcCombo ", ")}}{{if $funcOut}}{{if $funcIn}}, {{end}}error{{end}}]{{end}}()
+ } {{end}}{{end}}{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+{{end}}
+ default:
+ panic("Invalid signature for {{$upperName}}")
+ }
+ }
+ return nil
Review Comment:
We could get there if either the function doesn't have a StartBundle function (or whatever function we're dealing with here), or if we were unable to infer the types of the StartBundle function.
Returning nil in the first case is definitely the right thing, I updated to panic in the second case. The second should never happen, but I agree its better to be explicit
##########
sdks/go/pkg/beam/registration/doc.go:
##########
@@ -0,0 +1,31 @@
+// 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.
+
+//go:generate go install github.com/apache/beam/sdks/v2/go/cmd/specialize
+//go:generate specialize --package=registration --input=registration.tmpl --x=data,universals --imports=typex
+//go:generate go fmt
+
+/*
+Package registration contains functions for registering and optimizing your DoFn.
+
+This package contains generic registration/optimization function for each possible combination of input and output arities in a DoFn's ProcessElement function.
+For example, given a DoFn with a ProcessElement function that takes 4 inputs and returns 3 outputs, you can call
+register.DoFn4x3[input1 type, input2 type, input3 type, input4 type, output1 type, output2 type, output3 type](&doFn{}) during pipeline construction. This will
+register your DoFn and perform some reflection based optimization to significantly speed up execution at runtime.
+
+See DoFn2x1 for a full example.
+*/
+
Review Comment:
Neat - I did not understand this before
--
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] damccorm commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
damccorm commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1108594304
> but how do you feel about merging this after the 2.39 cut, so the first release that requires Go 1.18 is 2.40?
Another option (which I might like even more) is to review this PR as is, and then undo all the changes except for the template file and specialize/main.go. That would allow me to keep putting up my remaining generics changes for review, so that when we reach 2.40 all that needs to happen is running `go generate`, go.mod and CI changes, and adding back `registration_tests.go`.
That would save me some hard context shifts, but it may be more trouble than its worth - regardless, I'd prefer we prioritize finishing the [watermark reviews](https://github.com/apache/beam/pull/17374), and by the time those are all through 2.39 may have been cut (or be close to being cut)
--
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] asf-ci commented on pull request #17429: [BEAM-14347] Allow users to optimize DoFn execution with a single generic registration function
Posted by GitBox <gi...@apache.org>.
asf-ci commented on PR #17429:
URL: https://github.com/apache/beam/pull/17429#issuecomment-1105375135
Can one of the admins verify this patch?
--
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