sdks/go/pkg/beam/core/graph/bind.go - beam - Git at Google

 // 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 graph

 import (
 	"reflect"

 	"github.com/apache/beam/sdks/go/pkg/beam/core/funcx"
 	"github.com/apache/beam/sdks/go/pkg/beam/core/typex"
 	"github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
 )

 // TODO(herohde) 4/21/2017: Bind is where most user mistakes will likely show
 // up. We should verify that common mistakes yield reasonable errors.

 // Bind returns the inbound, outbound and underlying output types for a Fn,
 // when bound to the underlying input types. The complication of bind is
 // primarily that UserFns have loose signatures and bind must produce valid
 // type information for the execution plan.
 //
 // For example,
 //
 //     func (t EventTime, k typex.X, v int, emit func(string, typex.X))
 // or
 //     func (context.Context, k typex.X, v int) (string, typex.X, error)
 //
 // are UserFns that may take one or two incoming fulltypes: either KV<X,int>
 // or X with a singleton side input of type int. For the purpose of the
 // shape of data processing, the two forms are equivalent. The non-data types,
 // context.Context and error, are not part of the data signature, but in play
 // only at runtime.
 //
 // If either was bound to the input type [KV<string,int>], bind would return:
 //
 //     inbound:  [Main: KV<X,int>]
 //     outbound: [KV<string,X>]
 //     output:   [KV<string,string>]
 //
 // Note that it propagates the assignment of X to string in the output type.
 //
 // If either was instead bound to the input fulltypes [float, int], the
 // result would be:
 //
 //     inbound:  [Main: X, Singleton: int]
 //     outbound: [KV<string,X>]
 //     output:   [KV<string, float>]
 //
 // Here, the inbound shape and output types are different from before.
 func Bind(fn *funcx.Fn, typedefs map[string]reflect.Type, in ...typex.FullType) ([]typex.FullType, []InputKind, []typex.FullType, []typex.FullType, error) {
 	addContext := func(err error, fn *funcx.Fn) error {
 		return errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
 	}

 	inbound, kinds, err := findInbound(fn, in...)
 	if err != nil {
 		return nil, nil, nil, nil, addContext(err, fn)
 	}
 	outbound, err := findOutbound(fn)
 	if err != nil {
 		return nil, nil, nil, nil, addContext(err, fn)
 	}

 	subst, err := typex.Bind(inbound, in)
 	if err != nil {
 		return nil, nil, nil, nil, addContext(err, fn)
 	}
 	for k, v := range typedefs {
 		if substK, exists := subst[k]; exists {
 			err := errors.Errorf("cannot substitute type %v with %v, already defined as %v", k, v, substK)
 			return nil, nil, nil, nil, addContext(err, fn)
 		}
 		subst[k] = v
 	}

 	out, err := typex.Substitute(outbound, subst)
 	if err != nil {
 		return nil, nil, nil, nil, addContext(err, fn)
 	}
 	return inbound, kinds, outbound, out, nil
 }

 func findOutbound(fn *funcx.Fn) ([]typex.FullType, error) {
 	ret := trimIllegal(returnTypes(funcx.SubReturns(fn.Ret, fn.Returns(funcx.RetValue)...)))
 	params := funcx.SubParams(fn.Param, fn.Params(funcx.FnEmit)...)

 	var outbound []typex.FullType

 	// The direct output is the "main" output, if any.
 	switch len(ret) {
 	case 0:
 		break // ok: no direct output.
 	case 1:
 		outbound = append(outbound, typex.New(ret[0]))
 	case 2:
 		outbound = append(outbound, typex.NewKV(typex.New(ret[0]), typex.New(ret[1])))
 	default:
 		return nil, errors.Errorf("too many return values: %v", ret)
 	}

 	for _, param := range params {
 		values, _ := funcx.UnfoldEmit(param.T)
 		trimmed := trimIllegal(values)
 		if len(trimmed) == 2 {
 			outbound = append(outbound, typex.NewKV(typex.New(trimmed[0]), typex.New(trimmed[1])))
 		} else {
 			outbound = append(outbound, typex.New(trimmed[0]))
 		}
 	}
 	return outbound, nil
 }

 func returnTypes(list []funcx.ReturnParam) []reflect.Type {
 	var ret []reflect.Type
 	for _, elm := range list {
 		ret = append(ret, elm.T)
 	}
 	return ret
 }

 func findInbound(fn *funcx.Fn, in ...typex.FullType) ([]typex.FullType, []InputKind, error) {
 	// log.Printf("Bind inbound: %v %v", fn, in)
 	addContext := func(err error, p []funcx.FnParam, in interface{}) error {
 		return errors.WithContextf(err, "binding params %v to input %v", p, in)
 	}

 	var inbound []typex.FullType
 	var kinds []InputKind
 	params := funcx.SubParams(fn.Param, fn.Params(funcx.FnValue|funcx.FnIter|funcx.FnReIter)...)
 	index := 0
 	for _, input := range in {
 		arity, err := inboundArity(input, index == 0)
 		if err != nil {
 			return nil, nil, addContext(err, params, input)
 		}
 		if len(params)-index < arity {
 			return nil, nil, addContext(errors.New("too few params"), params[index:], input)
 		}

 		paramsToBind := params[index : index+arity]
 		elm, kind, err := tryBindInbound(input, paramsToBind, index == 0)
 		if err != nil {
 			return nil, nil, addContext(err, paramsToBind, input)
 		}
 		inbound = append(inbound, elm)
 		kinds = append(kinds, kind)
 		index += arity
 	}
 	if index < len(params) {
 		return nil, nil, addContext(errors.New("too few inputs: forgot an input or to annotate options?"), params, in)
 	}
 	if index > len(params) {
 		return nil, nil, addContext(errors.New("too many inputs"), params, in)
 	}
 	return inbound, kinds, nil
 }

 func tryBindInbound(t typex.FullType, args []funcx.FnParam, isMain bool) (typex.FullType, InputKind, error) {
 	kind := Main
 	var other typex.FullType

 	switch t.Class() {
 	case typex.Concrete, typex.Container:
 		if isMain {
 			other = typex.New(args[0].T)
 		} else {
 			// We accept various forms for side input. We have to disambiguate
 			// []string into a Singleton of type []string or a Slice of type
 			// string by matching up the incoming type and the param type.

 			arg := args[0]
 			switch arg.Kind {
 			case funcx.FnValue:
 				if args[0].T.Kind() == reflect.Slice && t.Type() == args[0].T.Elem() {
 					// TODO(herohde) 6/29/2017: we do not allow universal slices, for now.

 					kind = Slice
 					other = typex.New(args[0].T.Elem())
 				} else {
 					kind = Singleton
 					other = typex.New(args[0].T)
 				}
 			case funcx.FnIter:
 				values, _ := funcx.UnfoldIter(args[0].T)
 				trimmed := trimIllegal(values)
 				if len(trimmed) != 1 {
 					return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
 				}

 				kind = Iter
 				other = typex.New(trimmed[0])

 			case funcx.FnReIter:
 				values, _ := funcx.UnfoldReIter(args[0].T)
 				trimmed := trimIllegal(values)
 				if len(trimmed) != 1 {
 					return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
 				}

 				kind = ReIter
 				other = typex.New(trimmed[0])

 			default:
 				return nil, kind, errors.Errorf("unexpected param kind: %v", arg)
 			}
 		}
 	case typex.Composite:
 		switch t.Type() {
 		case typex.KVType:
 			if isMain {
 				if args[0].Kind != funcx.FnValue {
 					return nil, kind, errors.Errorf("key of %v cannot bind to %v", t, args[0])
 				}
 				if args[1].Kind != funcx.FnValue {
 					return nil, kind, errors.Errorf("value of %v cannot bind to %v", t, args[1])
 				}
 				other = typex.NewKV(typex.New(args[0].T), typex.New(args[1].T))
 			} else {
 				// TODO(herohde) 6/29/2017: side input map form.

 				switch args[0].Kind {
 				case funcx.FnIter:
 					values, _ := funcx.UnfoldIter(args[0].T)
 					trimmed := trimIllegal(values)
 					if len(trimmed) != 2 {
 						return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
 					}

 					kind = Iter
 					other = typex.NewKV(typex.New(trimmed[0]), typex.New(trimmed[1]))

 				case funcx.FnReIter:
 					values, _ := funcx.UnfoldReIter(args[0].T)
 					trimmed := trimIllegal(values)
 					if len(trimmed) != 2 {
 						return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
 					}

 					kind = ReIter
 					other = typex.NewKV(typex.New(trimmed[0]), typex.New(trimmed[1]))

 				default:
 					return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
 				}
 			}

 		case typex.CoGBKType:
 			if args[0].Kind != funcx.FnValue {
 				return nil, kind, errors.Errorf("key of %v cannot bind to %v", t, args[0])
 			}

 			components := []typex.FullType{typex.New(args[0].T)}

 			for i := 1; i < len(args); i++ {
 				switch args[i].Kind {
 				case funcx.FnIter:
 					values, _ := funcx.UnfoldIter(args[i].T)
 					trimmed := trimIllegal(values)
 					if len(trimmed) != 1 {
 						return nil, kind, errors.Errorf("values of %v cannot bind to %v", t, args[i])
 					}
 					components = append(components, typex.New(trimmed[0]))

 				case funcx.FnReIter:
 					values, _ := funcx.UnfoldReIter(args[i].T)
 					trimmed := trimIllegal(values)
 					if len(trimmed) != 1 {
 						return nil, kind, errors.Errorf("values of %v cannot bind to %v", t, args[i])
 					}
 					components = append(components, typex.New(trimmed[0]))
 				default:
 					return nil, kind, errors.Errorf("values of %v cannot bind to %v", t, args[i])
 				}
 			}
 			other = typex.NewCoGBK(components...)

 		default:
 			return nil, kind, errors.Errorf("unexpected inbound type: %v", t.Type())
 		}

 	default:
 		return nil, kind, errors.Errorf("unexpected inbound class: %v", t.Class())
 	}

 	if !typex.IsStructurallyAssignable(t, other) {
 		return nil, kind, errors.Errorf("%v is not assignable to %v", t, other)
 	}
 	return other, kind, nil
 }

 func inboundArity(t typex.FullType, isMain bool) (int, error) {
 	if t.Class() == typex.Composite {
 		switch t.Type() {
 		case typex.KVType:
 			if isMain {
 				return 2, nil
 			}
 			// A KV side input must be a single iterator/map.
 			return 1, nil
 		case typex.CoGBKType:
 			return len(t.Components()), nil
 		default:
 			return 0, errors.Errorf("unexpected composite inbound type: %v", t.Type())
 		}
 	}
 	return 1, nil
 }

 func trimIllegal(list []reflect.Type) []reflect.Type {
 	var ret []reflect.Type
 	for _, elm := range list {
 		switch typex.ClassOf(elm) {
 		case typex.Concrete, typex.Universal, typex.Container:
 			ret = append(ret, elm)
 		}
 	}
 	return ret
 }
	// 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 graph

	import (
	"reflect"

	"github.com/apache/beam/sdks/go/pkg/beam/core/funcx"
	"github.com/apache/beam/sdks/go/pkg/beam/core/typex"
	"github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
	)

	// TODO(herohde) 4/21/2017: Bind is where most user mistakes will likely show
	// up. We should verify that common mistakes yield reasonable errors.

	// Bind returns the inbound, outbound and underlying output types for a Fn,
	// when bound to the underlying input types. The complication of bind is
	// primarily that UserFns have loose signatures and bind must produce valid
	// type information for the execution plan.
	//
	// For example,
	//
	// func (t EventTime, k typex.X, v int, emit func(string, typex.X))
	// or
	// func (context.Context, k typex.X, v int) (string, typex.X, error)
	//
	// are UserFns that may take one or two incoming fulltypes: either KV<X,int>
	// or X with a singleton side input of type int. For the purpose of the
	// shape of data processing, the two forms are equivalent. The non-data types,
	// context.Context and error, are not part of the data signature, but in play
	// only at runtime.
	//
	// If either was bound to the input type [KV<string,int>], bind would return:
	//
	// inbound: [Main: KV<X,int>]
	// outbound: [KV<string,X>]
	// output: [KV<string,string>]
	//
	// Note that it propagates the assignment of X to string in the output type.
	//
	// If either was instead bound to the input fulltypes [float, int], the
	// result would be:
	//
	// inbound: [Main: X, Singleton: int]
	// outbound: [KV<string,X>]
	// output: [KV<string, float>]
	//
	// Here, the inbound shape and output types are different from before.
	func Bind(fn *funcx.Fn, typedefs map[string]reflect.Type, in ...typex.FullType) ([]typex.FullType, []InputKind, []typex.FullType, []typex.FullType, error) {
	addContext := func(err error, fn *funcx.Fn) error {
	return errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
	}

	inbound, kinds, err := findInbound(fn, in...)
	if err != nil {
	return nil, nil, nil, nil, addContext(err, fn)
	}
	outbound, err := findOutbound(fn)
	if err != nil {
	return nil, nil, nil, nil, addContext(err, fn)
	}

	subst, err := typex.Bind(inbound, in)
	if err != nil {
	return nil, nil, nil, nil, addContext(err, fn)
	}
	for k, v := range typedefs {
	if substK, exists := subst[k]; exists {
	err := errors.Errorf("cannot substitute type %v with %v, already defined as %v", k, v, substK)
	return nil, nil, nil, nil, addContext(err, fn)
	}
	subst[k] = v
	}

	out, err := typex.Substitute(outbound, subst)
	if err != nil {
	return nil, nil, nil, nil, addContext(err, fn)
	}
	return inbound, kinds, outbound, out, nil
	}

	func findOutbound(fn *funcx.Fn) ([]typex.FullType, error) {
	ret := trimIllegal(returnTypes(funcx.SubReturns(fn.Ret, fn.Returns(funcx.RetValue)...)))
	params := funcx.SubParams(fn.Param, fn.Params(funcx.FnEmit)...)

	var outbound []typex.FullType

	// The direct output is the "main" output, if any.
	switch len(ret) {
	case 0:
	break // ok: no direct output.
	case 1:
	outbound = append(outbound, typex.New(ret[0]))
	case 2:
	outbound = append(outbound, typex.NewKV(typex.New(ret[0]), typex.New(ret[1])))
	default:
	return nil, errors.Errorf("too many return values: %v", ret)
	}

	for _, param := range params {
	values, _ := funcx.UnfoldEmit(param.T)
	trimmed := trimIllegal(values)
	if len(trimmed) == 2 {
	outbound = append(outbound, typex.NewKV(typex.New(trimmed[0]), typex.New(trimmed[1])))
	} else {
	outbound = append(outbound, typex.New(trimmed[0]))
	}
	}
	return outbound, nil
	}

	func returnTypes(list []funcx.ReturnParam) []reflect.Type {
	var ret []reflect.Type
	for _, elm := range list {
	ret = append(ret, elm.T)
	}
	return ret
	}

	func findInbound(fn *funcx.Fn, in ...typex.FullType) ([]typex.FullType, []InputKind, error) {
	// log.Printf("Bind inbound: %v %v", fn, in)
	addContext := func(err error, p []funcx.FnParam, in interface{}) error {
	return errors.WithContextf(err, "binding params %v to input %v", p, in)
	}

	var inbound []typex.FullType
	var kinds []InputKind
	params := funcx.SubParams(fn.Param, fn.Params(funcx.FnValue\|funcx.FnIter\|funcx.FnReIter)...)
	index := 0
	for _, input := range in {
	arity, err := inboundArity(input, index == 0)
	if err != nil {
	return nil, nil, addContext(err, params, input)
	}
	if len(params)-index < arity {
	return nil, nil, addContext(errors.New("too few params"), params[index:], input)
	}

	paramsToBind := params[index : index+arity]
	elm, kind, err := tryBindInbound(input, paramsToBind, index == 0)
	if err != nil {
	return nil, nil, addContext(err, paramsToBind, input)
	}
	inbound = append(inbound, elm)
	kinds = append(kinds, kind)
	index += arity
	}
	if index < len(params) {
	return nil, nil, addContext(errors.New("too few inputs: forgot an input or to annotate options?"), params, in)
	}
	if index > len(params) {
	return nil, nil, addContext(errors.New("too many inputs"), params, in)
	}
	return inbound, kinds, nil
	}

	func tryBindInbound(t typex.FullType, args []funcx.FnParam, isMain bool) (typex.FullType, InputKind, error) {
	kind := Main
	var other typex.FullType

	switch t.Class() {
	case typex.Concrete, typex.Container:
	if isMain {
	other = typex.New(args[0].T)
	} else {
	// We accept various forms for side input. We have to disambiguate
	// []string into a Singleton of type []string or a Slice of type
	// string by matching up the incoming type and the param type.

	arg := args[0]
	switch arg.Kind {
	case funcx.FnValue:
	if args[0].T.Kind() == reflect.Slice && t.Type() == args[0].T.Elem() {
	// TODO(herohde) 6/29/2017: we do not allow universal slices, for now.

	kind = Slice
	other = typex.New(args[0].T.Elem())
	} else {
	kind = Singleton
	other = typex.New(args[0].T)
	}
	case funcx.FnIter:
	values, _ := funcx.UnfoldIter(args[0].T)
	trimmed := trimIllegal(values)
	if len(trimmed) != 1 {
	return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
	}

	kind = Iter
	other = typex.New(trimmed[0])

	case funcx.FnReIter:
	values, _ := funcx.UnfoldReIter(args[0].T)
	trimmed := trimIllegal(values)
	if len(trimmed) != 1 {
	return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
	}

	kind = ReIter
	other = typex.New(trimmed[0])

	default:
	return nil, kind, errors.Errorf("unexpected param kind: %v", arg)
	}
	}
	case typex.Composite:
	switch t.Type() {
	case typex.KVType:
	if isMain {
	if args[0].Kind != funcx.FnValue {
	return nil, kind, errors.Errorf("key of %v cannot bind to %v", t, args[0])
	}
	if args[1].Kind != funcx.FnValue {
	return nil, kind, errors.Errorf("value of %v cannot bind to %v", t, args[1])
	}
	other = typex.NewKV(typex.New(args[0].T), typex.New(args[1].T))
	} else {
	// TODO(herohde) 6/29/2017: side input map form.

	switch args[0].Kind {
	case funcx.FnIter:
	values, _ := funcx.UnfoldIter(args[0].T)
	trimmed := trimIllegal(values)
	if len(trimmed) != 2 {
	return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
	}

	kind = Iter
	other = typex.NewKV(typex.New(trimmed[0]), typex.New(trimmed[1]))

	case funcx.FnReIter:
	values, _ := funcx.UnfoldReIter(args[0].T)
	trimmed := trimIllegal(values)
	if len(trimmed) != 2 {
	return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
	}

	kind = ReIter
	other = typex.NewKV(typex.New(trimmed[0]), typex.New(trimmed[1]))

	default:
	return nil, kind, errors.Errorf("%v cannot bind to %v", t, args[0])
	}
	}

	case typex.CoGBKType:
	if args[0].Kind != funcx.FnValue {
	return nil, kind, errors.Errorf("key of %v cannot bind to %v", t, args[0])
	}

	components := []typex.FullType{typex.New(args[0].T)}

	for i := 1; i < len(args); i++ {
	switch args[i].Kind {
	case funcx.FnIter:
	values, _ := funcx.UnfoldIter(args[i].T)
	trimmed := trimIllegal(values)
	if len(trimmed) != 1 {
	return nil, kind, errors.Errorf("values of %v cannot bind to %v", t, args[i])
	}
	components = append(components, typex.New(trimmed[0]))

	case funcx.FnReIter:
	values, _ := funcx.UnfoldReIter(args[i].T)
	trimmed := trimIllegal(values)
	if len(trimmed) != 1 {
	return nil, kind, errors.Errorf("values of %v cannot bind to %v", t, args[i])
	}
	components = append(components, typex.New(trimmed[0]))
	default:
	return nil, kind, errors.Errorf("values of %v cannot bind to %v", t, args[i])
	}
	}
	other = typex.NewCoGBK(components...)

	default:
	return nil, kind, errors.Errorf("unexpected inbound type: %v", t.Type())
	}

	default:
	return nil, kind, errors.Errorf("unexpected inbound class: %v", t.Class())
	}

	if !typex.IsStructurallyAssignable(t, other) {
	return nil, kind, errors.Errorf("%v is not assignable to %v", t, other)
	}
	return other, kind, nil
	}

	func inboundArity(t typex.FullType, isMain bool) (int, error) {
	if t.Class() == typex.Composite {
	switch t.Type() {
	case typex.KVType:
	if isMain {
	return 2, nil
	}
	// A KV side input must be a single iterator/map.
	return 1, nil
	case typex.CoGBKType:
	return len(t.Components()), nil
	default:
	return 0, errors.Errorf("unexpected composite inbound type: %v", t.Type())
	}
	}
	return 1, nil
	}

	func trimIllegal(list []reflect.Type) []reflect.Type {
	var ret []reflect.Type
	for _, elm := range list {
	switch typex.ClassOf(elm) {
	case typex.Concrete, typex.Universal, typex.Container:
	ret = append(ret, elm)
	}
	}
	return ret
	}