You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2022/05/09 18:49:07 UTC
[beam] branch master updated: [BEAM-14347] Add generic registration functions for iters and emitters (#17574)
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 36d40f40694 [BEAM-14347] Add generic registration functions for iters and emitters (#17574)
36d40f40694 is described below
commit 36d40f40694395dc346192ab1c71b5a9e79daa03
Author: Danny McCormick <da...@google.com>
AuthorDate: Mon May 9 14:49:02 2022 -0400
[BEAM-14347] Add generic registration functions for iters and emitters (#17574)
---
sdks/go/pkg/beam/registration/emitter.go | 164 +++++++++++++++
.../beam/registration/example_registration_test.go | 43 +++-
sdks/go/pkg/beam/registration/iter.go | 129 ++++++++++++
sdks/go/pkg/beam/registration/registration_test.go | 226 +++++++++++++++++++++
4 files changed, 557 insertions(+), 5 deletions(-)
diff --git a/sdks/go/pkg/beam/registration/emitter.go b/sdks/go/pkg/beam/registration/emitter.go
new file mode 100644
index 00000000000..347aa641272
--- /dev/null
+++ b/sdks/go/pkg/beam/registration/emitter.go
@@ -0,0 +1,164 @@
+// 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
+
+import (
+ "context"
+ "reflect"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
+)
+
+type emit struct {
+ est *sdf.WatermarkEstimator
+
+ ctx context.Context
+ ws []typex.Window
+ et typex.EventTime
+ value exec.FullValue
+}
+
+func (e *emit) Init(ctx context.Context, ws []typex.Window, et typex.EventTime) error {
+ e.ctx = ctx
+ e.ws = ws
+ e.et = et
+ return nil
+}
+
+func (e *emit) AttachEstimator(est *sdf.WatermarkEstimator) {
+ e.est = est
+}
+
+type emit1[T any] struct {
+ emit
+ n exec.ElementProcessor
+}
+
+func (e *emit1[T]) Value() interface{} {
+ return e.invoke
+}
+
+func (e *emit1[T]) invoke(val T) {
+ e.value = exec.FullValue{Windows: e.ws, Timestamp: e.et, Elm: val}
+ if e.est != nil {
+ (*e.est).(sdf.TimestampObservingEstimator).ObserveTimestamp(e.et.ToTime())
+ }
+ if err := e.n.ProcessElement(e.ctx, &e.value); err != nil {
+ panic(err)
+ }
+}
+
+type emit2[T1, T2 any] struct {
+ emit
+ n exec.ElementProcessor
+}
+
+func (e *emit2[T1, T2]) Value() interface{} {
+ return e.invoke
+}
+
+func (e *emit2[T1, T2]) invoke(key T1, val T2) {
+ e.value = exec.FullValue{Windows: e.ws, Timestamp: e.et, Elm: key, Elm2: val}
+ if e.est != nil {
+ (*e.est).(sdf.TimestampObservingEstimator).ObserveTimestamp(e.et.ToTime())
+ }
+ if err := e.n.ProcessElement(e.ctx, &e.value); err != nil {
+ panic(err)
+ }
+}
+
+type emit1WithTimestamp[T any] struct {
+ emit
+ n exec.ElementProcessor
+}
+
+func (e *emit1WithTimestamp[T]) Value() interface{} {
+ return e.invoke
+}
+
+func (e *emit1WithTimestamp[T]) invoke(et typex.EventTime, val T) {
+ e.value = exec.FullValue{Windows: e.ws, Timestamp: et, Elm: val}
+ if e.est != nil {
+ (*e.est).(sdf.TimestampObservingEstimator).ObserveTimestamp(et.ToTime())
+ }
+ if err := e.n.ProcessElement(e.ctx, &e.value); err != nil {
+ panic(err)
+ }
+}
+
+type emit2WithTimestamp[T1, T2 any] struct {
+ emit
+ n exec.ElementProcessor
+}
+
+func (e *emit2WithTimestamp[T1, T2]) Value() interface{} {
+ return e.invoke
+}
+
+func (e *emit2WithTimestamp[T1, T2]) invoke(et typex.EventTime, key T1, val T2) {
+ e.value = exec.FullValue{Windows: e.ws, Timestamp: et, Elm: key, Elm2: val}
+ if e.est != nil {
+ (*e.est).(sdf.TimestampObservingEstimator).ObserveTimestamp(et.ToTime())
+ }
+ if err := e.n.ProcessElement(e.ctx, &e.value); err != nil {
+ panic(err)
+ }
+}
+
+// Emitter1 registers parameters from your DoFn with a
+// signature func(T) and optimizes their execution.
+// This must be done by passing in type parameters of your input as a constraint,
+// aka: registration.Emitter1[T]()
+func Emitter1[T1 any]() {
+ e := (*func(T1))(nil)
+ registerFunc := func(n exec.ElementProcessor) exec.ReusableEmitter {
+ return &emit1[T1]{n: n}
+ }
+ exec.RegisterEmitter(reflect.TypeOf(e).Elem(), registerFunc)
+}
+
+// Emitter2 registers parameters from your DoFn with a
+// signature func(T1, T2) and optimizes their execution.
+// This must be done by passing in type parameters of all inputs (including EventTime)
+// as constraints, aka: registration.Emitter2[T1, T2](), where T2 is the type of your
+// value and T2 is either the type of your key or the eventTime.
+func Emitter2[T1, T2 any]() {
+ e := (*func(T1, T2))(nil)
+ registerFunc := func(n exec.ElementProcessor) exec.ReusableEmitter {
+ return &emit2[T1, T2]{n: n}
+ }
+ if reflect.TypeOf(e).Elem().In(0) == typex.EventTimeType {
+ registerFunc = func(n exec.ElementProcessor) exec.ReusableEmitter {
+ return &emit1WithTimestamp[T2]{n: n}
+ }
+ }
+ exec.RegisterEmitter(reflect.TypeOf(e).Elem(), registerFunc)
+}
+
+// Emitter3 registers parameters from your DoFn with a
+// signature func(T1, T2, T3) and optimizes their execution.
+// This must be done by passing in type parameters of all inputs as constraints,
+// aka: registration.Emitter3[beam.EventTime, T1, T2](), where T1 is the type of
+// your key and T2 is the type of your value.
+func Emitter3[T1 typex.EventTime, T2, T3 any]() {
+ e := (*func(T1, T2, T3))(nil)
+ registerFunc := func(n exec.ElementProcessor) exec.ReusableEmitter {
+ return &emit2WithTimestamp[T2, T3]{n: n}
+ }
+ exec.RegisterEmitter(reflect.TypeOf(e).Elem(), registerFunc)
+}
diff --git a/sdks/go/pkg/beam/registration/example_registration_test.go b/sdks/go/pkg/beam/registration/example_registration_test.go
index 8ac5eba62b8..9392e758984 100644
--- a/sdks/go/pkg/beam/registration/example_registration_test.go
+++ b/sdks/go/pkg/beam/registration/example_registration_test.go
@@ -18,14 +18,18 @@ package registration_test
import (
"context"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/v2/go/pkg/beam/registration"
)
-type myDoFn struct {
-}
+type myDoFn struct{}
-func (fn *myDoFn) ProcessElement(word string, emit func(int)) int {
- emit(len(word))
+func (fn *myDoFn) ProcessElement(word string, iter func(*string) bool, emit func(int)) int {
+ var s string
+ for iter(&s) {
+ emit(len(s))
+ }
return len(word)
}
@@ -46,9 +50,38 @@ func (fn *myDoFn) Teardown() error {
return nil
}
+type myDoFn2 struct{}
+
+type Foo struct {
+ s string
+}
+
+func (fn *myDoFn2) ProcessElement(word string, iter func(**Foo, *beam.EventTime) bool, emit func(beam.EventTime, string, int)) (beam.EventTime, string, int) {
+ var f *Foo
+ var et beam.EventTime
+ for iter(&f, &et) {
+ emit(et, f.s, len(f.s))
+ }
+ return mtime.Now(), word, len(word)
+}
+
func ExampleDoFn2x1() {
// Since myDoFn's ProcessElement call has 2 inputs and 1 output, call DoFn2x1.
// Since the inputs to ProcessElement are (string, func(int)), and the output
// is int, we pass those parameter types to the function.
- registration.DoFn2x1[string, func(int), int](&myDoFn{})
+ registration.DoFn3x1[string, func(*string) bool, func(int), int](&myDoFn{})
+
+ // Any function parameters (iters or emitters) must be registered separately
+ // as well to get the fully optimized experience. Since ProcessElement has
+ // an emitter with the signature func(int) we can register it. This must be
+ // done by passing in the type parameters of all inputs as constraints.
+ registration.Emitter1[int]()
+ registration.Iter1[string]()
+
+ registration.DoFn3x3[string, func(**Foo, *beam.EventTime) bool, func(beam.EventTime, string, int), beam.EventTime, string, int](&myDoFn2{})
+
+ // More complex iter/emitter registration work in the same way, even when
+ // timestamps or pointers are involved.
+ registration.Emitter3[beam.EventTime, string, int]()
+ registration.Iter2[*Foo, beam.EventTime]()
}
diff --git a/sdks/go/pkg/beam/registration/iter.go b/sdks/go/pkg/beam/registration/iter.go
new file mode 100644
index 00000000000..4a1e69b4dba
--- /dev/null
+++ b/sdks/go/pkg/beam/registration/iter.go
@@ -0,0 +1,129 @@
+// 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
+
+import (
+ "fmt"
+ "io"
+ "reflect"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
+)
+
+type iter1[T any] struct {
+ s exec.ReStream
+
+ // cur is the "current" stream, if any.
+ cur exec.Stream
+}
+
+func (v *iter1[T]) Init() error {
+ cur, err := v.s.Open()
+ if err != nil {
+ return err
+ }
+ v.cur = cur
+ return nil
+}
+
+func (v *iter1[T]) Value() interface{} {
+ return v.invoke
+}
+
+func (v *iter1[T]) Reset() error {
+ if err := v.cur.Close(); err != nil {
+ return err
+ }
+ v.cur = nil
+ return nil
+}
+
+func (v *iter1[T]) invoke(value *T) bool {
+ elm, err := v.cur.Read()
+ if err != nil {
+ if err == io.EOF {
+ return false
+ }
+ panic(fmt.Sprintf("broken stream: %v", err))
+ }
+ *value = elm.Elm.(T)
+ return true
+}
+
+type iter2[T1, T2 any] struct {
+ s exec.ReStream
+
+ // cur is the "current" stream, if any.
+ cur exec.Stream
+}
+
+func (v *iter2[T1, T2]) Init() error {
+ cur, err := v.s.Open()
+ if err != nil {
+ return err
+ }
+ v.cur = cur
+ return nil
+}
+
+func (v *iter2[T1, T2]) Value() interface{} {
+ return v.invoke
+}
+
+func (v *iter2[T1, T2]) Reset() error {
+ if err := v.cur.Close(); err != nil {
+ return err
+ }
+ v.cur = nil
+ return nil
+}
+
+func (v *iter2[T1, T2]) invoke(key *T1, value *T2) bool {
+ elm, err := v.cur.Read()
+ if err != nil {
+ if err == io.EOF {
+ return false
+ }
+ panic(fmt.Sprintf("broken stream: %v", err))
+ }
+ *key = elm.Elm.(T1)
+ *value = elm.Elm2.(T2)
+ return true
+}
+
+// Iter1 registers parameters from your DoFn with a
+// signature func(*T) bool and optimizes their execution.
+// This must be done by passing in type parameters of all inputs as constraints,
+// aka: registration.Iter1[T]()
+func Iter1[T any]() {
+ i := (*func(*T) bool)(nil)
+ registerFunc := func(s exec.ReStream) exec.ReusableInput {
+ return &iter1[T]{s: s}
+ }
+ exec.RegisterInput(reflect.TypeOf(i).Elem(), registerFunc)
+}
+
+// Iter1 registers parameters from your DoFn with a
+// signature func(*T1, *T2) bool and optimizes their execution.
+// This must be done by passing in type parameters of all inputs as constraints,
+// aka: registration.Iter2[T1, T2]()
+func Iter2[T1, T2 any]() {
+ i := (*func(*T1, *T2) bool)(nil)
+ registerFunc := func(s exec.ReStream) exec.ReusableInput {
+ return &iter2[T1, T2]{s: s}
+ }
+ exec.RegisterInput(reflect.TypeOf(i).Elem(), registerFunc)
+}
diff --git a/sdks/go/pkg/beam/registration/registration_test.go b/sdks/go/pkg/beam/registration/registration_test.go
index 814c8bc6d3c..4f5c37dd2a8 100644
--- a/sdks/go/pkg/beam/registration/registration_test.go
+++ b/sdks/go/pkg/beam/registration/registration_test.go
@@ -20,7 +20,11 @@ import (
"reflect"
"testing"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
"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"
)
@@ -155,6 +159,228 @@ func TestRegister_RegistersTypes(t *testing.T) {
}
}
+func TestEmitter1(t *testing.T) {
+ Emitter1[int]()
+ if !exec.IsEmitterRegistered(reflect.TypeOf((*func(int))(nil)).Elem()) {
+ t.Fatalf("exec.IsEmitterRegistered(reflect.TypeOf((*func(int))(nil)).Elem()) = false, want true")
+ }
+}
+
+func TestEmitter2(t *testing.T) {
+ Emitter2[int, string]()
+ if !exec.IsEmitterRegistered(reflect.TypeOf((*func(int, string))(nil)).Elem()) {
+ t.Fatalf("exec.IsEmitterRegistered(reflect.TypeOf((*func(int, string))(nil)).Elem()) = false, want true")
+ }
+}
+
+func TestEmitter2_WithTimestamp(t *testing.T) {
+ Emitter2[typex.EventTime, string]()
+ if !exec.IsEmitterRegistered(reflect.TypeOf((*func(typex.EventTime, string))(nil)).Elem()) {
+ t.Fatalf("exec.IsEmitterRegistered(reflect.TypeOf((*func(typex.EventTime, string))(nil)).Elem()) = false, want true")
+ }
+}
+
+func TestEmitter3(t *testing.T) {
+ Emitter3[typex.EventTime, int, string]()
+ if !exec.IsEmitterRegistered(reflect.TypeOf((*func(typex.EventTime, int, string))(nil)).Elem()) {
+ t.Fatalf("exec.IsEmitterRegistered(reflect.TypeOf((*func(typex.EventTime, int, string))(nil)).Elem()) = false, want true")
+ }
+}
+
+func TestEmit1(t *testing.T) {
+ e := &emit1[int]{n: &elementProcessor{}}
+ e.Init(context.Background(), []typex.Window{}, mtime.ZeroTimestamp)
+ fn := e.Value().(func(int))
+ fn(3)
+ if got, want := e.n.(*elementProcessor).inFV.Elm, 3; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm=%v, want %v", got, want)
+ }
+ if got := e.n.(*elementProcessor).inFV.Elm2; got != nil {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm2=%v, want nil", got)
+ }
+ if got, want := e.n.(*elementProcessor).inFV.Timestamp, mtime.ZeroTimestamp; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Timestamp=%v, want %v", got, want)
+ }
+}
+
+func TestEmit2(t *testing.T) {
+ e := &emit2[int, string]{n: &elementProcessor{}}
+ e.Init(context.Background(), []typex.Window{}, mtime.ZeroTimestamp)
+ fn := e.Value().(func(int, string))
+ fn(3, "hello")
+ if got, want := e.n.(*elementProcessor).inFV.Elm, 3; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm=%v, want %v", got, want)
+ }
+ if got, want := e.n.(*elementProcessor).inFV.Elm2, "hello"; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm2=%v, want %v", got, want)
+ }
+ if got, want := e.n.(*elementProcessor).inFV.Timestamp, mtime.ZeroTimestamp; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Timestamp=%v, want %v", got, want)
+ }
+}
+
+func TestEmit1WithTimestamp(t *testing.T) {
+ e := &emit1WithTimestamp[int]{n: &elementProcessor{}}
+ e.Init(context.Background(), []typex.Window{}, mtime.ZeroTimestamp)
+ fn := e.Value().(func(typex.EventTime, int))
+ fn(mtime.MaxTimestamp, 3)
+ if got, want := e.n.(*elementProcessor).inFV.Elm, 3; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm=%v, want %v", got, want)
+ }
+ if got := e.n.(*elementProcessor).inFV.Elm2; got != nil {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm2=%v, want nil", got)
+ }
+ if got, want := e.n.(*elementProcessor).inFV.Timestamp, mtime.MaxTimestamp; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Timestamp=%v, want %v", got, want)
+ }
+}
+
+func TestEmit2WithTimestamp(t *testing.T) {
+ e := &emit2WithTimestamp[int, string]{n: &elementProcessor{}}
+ e.Init(context.Background(), []typex.Window{}, mtime.ZeroTimestamp)
+ fn := e.Value().(func(typex.EventTime, int, string))
+ fn(mtime.MaxTimestamp, 3, "hello")
+ if got, want := e.n.(*elementProcessor).inFV.Elm, 3; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm=%v, want %v", got, want)
+ }
+ if got, want := e.n.(*elementProcessor).inFV.Elm2, "hello"; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Elm2=%v, want %v", got, want)
+ }
+ if got, want := e.n.(*elementProcessor).inFV.Timestamp, mtime.MaxTimestamp; got != want {
+ t.Errorf("e.Value.(func(int))(3).n.inFV.Timestamp=%v, want %v", got, want)
+ }
+}
+
+func TestIter1(t *testing.T) {
+ Iter1[int]()
+ if !exec.IsInputRegistered(reflect.TypeOf((*func(*int) bool)(nil)).Elem()) {
+ t.Fatalf("exec.IsInputRegistered(reflect.TypeOf(((*func(*int) bool)(nil)).Elem()) = false, want true")
+ }
+}
+
+func TestIter2(t *testing.T) {
+ Iter2[int, string]()
+ if !exec.IsInputRegistered(reflect.TypeOf((*func(*int, *string) bool)(nil)).Elem()) {
+ t.Fatalf("exec.IsInputRegistered(reflect.TypeOf((*func(*int, *string) bool)(nil)).Elem()) = false, want true")
+ }
+}
+
+func TestIter1_Struct(t *testing.T) {
+ values := []exec.FullValue{exec.FullValue{
+ Windows: window.SingleGlobalWindow,
+ Timestamp: mtime.ZeroTimestamp,
+ Elm: "one",
+ }, exec.FullValue{
+ Windows: window.SingleGlobalWindow,
+ Timestamp: mtime.ZeroTimestamp,
+ Elm: "two",
+ }, exec.FullValue{
+ Windows: window.SingleGlobalWindow,
+ Timestamp: mtime.ZeroTimestamp,
+ Elm: "three",
+ }}
+
+ i := iter1[string]{s: &exec.FixedReStream{Buf: values}}
+
+ i.Init()
+ fn := i.Value().(func(value *string) bool)
+
+ var s string
+ if ok := fn(&s); !ok {
+ t.Fatalf("First i.Value()(&s)=false, want true")
+ }
+ if got, want := s, "one"; got != want {
+ t.Fatalf("First iter value = %v, want %v", got, want)
+ }
+ if ok := fn(&s); !ok {
+ t.Fatalf("Second i.Value()(&s)=false, want true")
+ }
+ if got, want := s, "two"; got != want {
+ t.Fatalf("Second iter value = %v, want %v", got, want)
+ }
+ if ok := fn(&s); !ok {
+ t.Fatalf("Third i.Value()(&s)=false, want true")
+ }
+ if got, want := s, "three"; got != want {
+ t.Fatalf("Third iter value = %v, want %v", got, want)
+ }
+ if ok := fn(&s); ok {
+ t.Fatalf("Fourth i.Value()(&s)=true, want false")
+ }
+ if err := i.Reset(); err != nil {
+ t.Fatalf("i.Reset()=%v, want nil", err)
+ }
+}
+
+func TestIter2_Struct(t *testing.T) {
+ values := []exec.FullValue{exec.FullValue{
+ Windows: window.SingleGlobalWindow,
+ Timestamp: mtime.ZeroTimestamp,
+ Elm: 1,
+ Elm2: "one",
+ }, exec.FullValue{
+ Windows: window.SingleGlobalWindow,
+ Timestamp: mtime.ZeroTimestamp,
+ Elm: 2,
+ Elm2: "two",
+ }, exec.FullValue{
+ Windows: window.SingleGlobalWindow,
+ Timestamp: mtime.ZeroTimestamp,
+ Elm: 3,
+ Elm2: "three",
+ }}
+
+ i := iter2[int, string]{s: &exec.FixedReStream{Buf: values}}
+
+ i.Init()
+ fn := i.Value().(func(key *int, value *string) bool)
+
+ var s string
+ var key int
+ if ok := fn(&key, &s); !ok {
+ t.Fatalf("First i.Value()(&s)=false, want true")
+ }
+ if got, want := key, 1; got != want {
+ t.Fatalf("First iter key = %v, want %v", got, want)
+ }
+ if got, want := s, "one"; got != want {
+ t.Fatalf("First iter value = %v, want %v", got, want)
+ }
+ if ok := fn(&key, &s); !ok {
+ t.Fatalf("Second i.Value()(&s)=false, want true")
+ }
+ if got, want := key, 2; got != want {
+ t.Fatalf("Second iter key = %v, want %v", got, want)
+ }
+ if got, want := s, "two"; got != want {
+ t.Fatalf("Second iter value = %v, want %v", got, want)
+ }
+ if ok := fn(&key, &s); !ok {
+ t.Fatalf("Third i.Value()(&s)=false, want true")
+ }
+ if got, want := key, 3; got != want {
+ t.Fatalf("Third iter key = %v, want %v", got, want)
+ }
+ if got, want := s, "three"; got != want {
+ t.Fatalf("Third iter value = %v, want %v", got, want)
+ }
+ if ok := fn(&key, &s); ok {
+ t.Fatalf("Fourth i.Value()(&s)=true, want false")
+ }
+ if err := i.Reset(); err != nil {
+ t.Fatalf("i.Reset()=%v, want nil", err)
+ }
+}
+
+type elementProcessor struct {
+ inFV exec.FullValue
+}
+
+func (e *elementProcessor) ProcessElement(ctx context.Context, elm *exec.FullValue, values ...exec.ReStream) error {
+ e.inFV = *elm
+ return nil
+}
+
type CompleteDoFn2x1 struct {
}