[
https://issues.apache.org/jira/browse/BEAM-14470?focusedWorklogId=777046&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-777046
]
ASF GitHub Bot logged work on BEAM-14470:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 01/Jun/22 17:36
Start Date: 01/Jun/22 17:36
Worklog Time Spent: 10m
Work Description: lostluck commented on code in PR #17790:
URL: https://github.com/apache/beam/pull/17790#discussion_r887129346
##########
sdks/go/pkg/beam/core/graph/fn_test.go:
##########
@@ -378,6 +380,129 @@ func TestNewCombineFn(t *testing.T) {
})
}
+func TestNewFn_DoFn(t *testing.T) {
+ // Validate wrap fallthrough
+ reflectx.RegisterStructWrapper(reflect.TypeOf((*GoodDoFn)(nil)).Elem(),
func(fn interface{}) map[string]reflectx.Func {
+ gdf := fn.(*GoodDoFn)
+ return map[string]reflectx.Func{
+ processElementName: reflectx.MakeFunc1x1(func(v int)
int {
+ return gdf.ProcessElement(v)
+ }),
+ }
+ })
+
+ userFn := &GoodDoFn{}
+ fn, err := NewFn(userFn)
+ if err != nil {
+ t.Errorf("NewFn(%T) failed:\n%v", userFn, err)
+ }
+ dofn, err := AsDoFn(fn, MainSingle)
+ if err != nil {
+ t.Errorf("AsDoFn(%v, MainSingle) failed:\n%v", fn.Name(), err)
+ }
+ // Check that we get expected values for all the methods.
+ if got, want := dofn.Name(), "GoodDoFn"; !strings.HasSuffix(got, want) {
+ t.Errorf("(%v).Name() = %q, want suffix %q", dofn.Name(), got,
want)
+ }
+ if dofn.SetupFn() == nil {
+ t.Errorf("(%v).SetupFn() == nil, want value", dofn.Name())
+ }
+ if dofn.StartBundleFn() == nil {
+ t.Errorf("(%v).StartBundleFn() == nil, want value", dofn.Name())
+ }
+ if dofn.ProcessElementFn() == nil {
+ t.Errorf("(%v).ProcessElementFn() == nil, want value",
dofn.Name())
+ }
+ if dofn.FinishBundleFn() == nil {
+ t.Errorf("(%v).FinishBundleFn() == nil, want value",
dofn.Name())
+ }
+ if dofn.TeardownFn() == nil {
+ t.Errorf("(%v).TeardownFn() == nil, want value", dofn.Name())
+ }
+ if dofn.IsSplittable() {
+ t.Errorf("(%v).IsSplittable() = true, want false", dofn.Name())
+ }
+}
+
+func TestNewFn_SplittableDoFn(t *testing.T) {
Review Comment:
Ah, I knew I missed something. Thank you!
We also have a vestigial "CompactFn" It's intended to allow CombineFns to do
compression of the accumulators, but it seems it's not documented or called
anywhere. Out of scope to delete it in this PR, but it could be worth
implementing for user use eventually.
Issue Time Tracking
-------------------
Worklog Id: (was: 777046)
Time Spent: 2.5h (was: 2h 20m)
> Optimize load tests w/generic registrations
> -------------------------------------------
>
> Key: BEAM-14470
> URL: https://issues.apache.org/jira/browse/BEAM-14470
> Project: Beam
> Issue Type: Improvement
> Components: sdk-go, testing
> Reporter: Robert Burke
> Assignee: Robert Burke
> Priority: P2
> Fix For: 2.40.0
>
> Time Spent: 2.5h
> Remaining Estimate: 0h
>
> With Generic Registration available, a way to indicate the value would be via
> the load tests.
> In particular the following tests should be optimized:
> http://metrics.beam.apache.org/d/fK0U4JqWz/cogbk-load-tests?orgId=1&var-processingType=batch&var-sdk=go
> http://metrics.beam.apache.org/d/WNzYt13Zk/combine-load-tests?orgId=1&var-processingType=batch&var-sdk=go
> http://metrics.beam.apache.org/d/UYZ-oJ3Zk/gbk-load-tests?orgId=1&var-processingType=batch&var-sdk=go
> http://metrics.beam.apache.org/d/MOi-kf3Zk/pardo-load-tests?orgId=1&var-processingType=batch&var-sdk=go
> http://metrics.beam.apache.org/d/-E9aGlFGk/side-input-load-tests?orgId=1&var-processingType=batch&var-sdk=go
> Benchmarks show reduce per call overhead by about ~300ns per parameter in the
> invoked user function, just like the code generated, and any improvement
> should be reflected by the load test graphs going forward.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
