[beam] branch master updated: [BEAM-7086] Refactoring repetitive error code.

[al...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

altay 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 aef01da  [BEAM-7086] Refactoring repetitive error code.
     new 5556467  Merge pull request #8674 from youngoli/beam7086
aef01da is described below

commit aef01da787f6c92bf8d7491dfa89ac43cbe85ace
Author: Daniel Oliveira <da...@gmail.com>
AuthorDate: Thu May 23 15:46:19 2019 -0700

    [BEAM-7086] Refactoring repetitive error code.
    
    Straightforward change, make some function literals for the really
    repetitive errors.
---
 sdks/go/pkg/beam/core/graph/bind.go | 30 ++++++++++++----------
 sdks/go/pkg/beam/core/graph/edge.go | 51 ++++++++++++++++++-------------------
 2 files changed, 42 insertions(+), 39 deletions(-)

diff --git a/sdks/go/pkg/beam/core/graph/bind.go b/sdks/go/pkg/beam/core/graph/bind.go
index 77a6d69..f5cc7f8 100644
--- a/sdks/go/pkg/beam/core/graph/bind.go
+++ b/sdks/go/pkg/beam/core/graph/bind.go
@@ -60,30 +60,34 @@ import (
 //
 // 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, errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
+		return nil, nil, nil, nil, addContext(err, fn)
 	}
 	outbound, err := findOutbound(fn)
 	if err != nil {
-		return nil, nil, nil, nil, errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
+		return nil, nil, nil, nil, addContext(err, fn)
 	}
 
 	subst, err := typex.Bind(inbound, in)
 	if err != nil {
-		return nil, nil, nil, nil, errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
+		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, errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
+			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, errors.WithContextf(err, "binding fn %v", fn.Fn.Name())
+		return nil, nil, nil, nil, addContext(err, fn)
 	}
 	return inbound, kinds, outbound, out, nil
 }
@@ -128,6 +132,9 @@ func returnTypes(list []funcx.ReturnParam) []reflect.Type {
 
 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
@@ -136,29 +143,26 @@ func findInbound(fn *funcx.Fn, in ...typex.FullType) ([]typex.FullType, []InputK
 	for _, input := range in {
 		arity, err := inboundArity(input, index == 0)
 		if err != nil {
-			return nil, nil, errors.WithContextf(err, "binding params %v to input %v", params, input)
+			return nil, nil, addContext(err, params, input)
 		}
 		if len(params)-index < arity {
-			err := errors.New("too few params")
-			return nil, nil, errors.WithContextf(err, "binding params %v to input %v", params[index:], input)
+			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, errors.WithContextf(err, "binding params %v to input %v", paramsToBind, input)
+			return nil, nil, addContext(err, paramsToBind, input)
 		}
 		inbound = append(inbound, elm)
 		kinds = append(kinds, kind)
 		index += arity
 	}
 	if index < len(params) {
-		err := errors.New("too few inputs: forgot an input or to annotate options?")
-		return nil, nil, errors.WithContextf(err, "binding params %v to inputs %v:", params, in)
+		return nil, nil, addContext(errors.New("too few inputs: forgot an input or to annotate options?"), params, in)
 	}
 	if index > len(params) {
-		err := errors.New("too many inputs")
-		return nil, nil, errors.WithContextf(err, "binding params %v to inputs %v:", params, in)
+		return nil, nil, addContext(errors.New("too many inputs"), params, in)
 	}
 	return inbound, kinds, nil
 }
diff --git a/sdks/go/pkg/beam/core/graph/edge.go b/sdks/go/pkg/beam/core/graph/edge.go
index 846dceb..dcdfd0b 100644
--- a/sdks/go/pkg/beam/core/graph/edge.go
+++ b/sdks/go/pkg/beam/core/graph/edge.go
@@ -186,14 +186,15 @@ func (e *MultiEdge) String() string {
 
 // NewCoGBK inserts a new CoGBK edge into the graph.
 func NewCoGBK(g *Graph, s *Scope, ns []*Node) (*MultiEdge, error) {
+	addContext := func(err error, s *Scope) error {
+		return errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+	}
+
 	if len(ns) == 0 {
-		// TODO(BEAM-7086) Reduce the repetition in the context of all the errors in this file.
-		err := errors.New("needs at least 1 input")
-		return nil, errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+		return nil, addContext(errors.New("needs at least 1 input"), s)
 	}
 	if !typex.IsKV(ns[0].Type()) {
-		err := errors.Errorf("input type must be KV: %v", ns[0])
-		return nil, errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+		return nil, addContext(errors.Errorf("input type must be KV: %v", ns[0]), s)
 	}
 
 	// (1) Create CoGBK result type: KV<T,U>, .., KV<T,Z> -> CoGBK<T,U,..,Z>.
@@ -206,20 +207,16 @@ func NewCoGBK(g *Graph, s *Scope, ns []*Node) (*MultiEdge, error) {
 	for i := 1; i < len(ns); i++ {
 		n := ns[i]
 		if !typex.IsKV(n.Type()) {
-			err := errors.Errorf("input type must be KV: %v", n)
-			return nil, errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+			return nil, addContext(errors.Errorf("input type must be KV: %v", n), s)
 		}
 		if !n.Coder.Components[0].Equals(c) {
-			err := errors.Errorf("key coder for %v is %v, want %v", n, n.Coder.Components[0], c)
-			return nil, errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+			return nil, addContext(errors.Errorf("key coder for %v is %v, want %v", n, n.Coder.Components[0], c), s)
 		}
 		if !w.Equals(n.WindowingStrategy()) {
-			err := errors.Errorf("mismatched CoGBK windowing strategies: %v, want %v", n.WindowingStrategy(), w)
-			return nil, errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+			return nil, addContext(errors.Errorf("mismatched CoGBK windowing strategies: %v, want %v", n.WindowingStrategy(), w), s)
 		}
 		if bounded != n.Bounded() {
-			err := errors.Errorf("unmatched CoGBK boundedness: %v, want %v", n.Bounded(), bounded)
-			return nil, errors.WithContextf(err, "creating new CoGBK in scope %v", s)
+			return nil, addContext(errors.Errorf("unmatched CoGBK boundedness: %v, want %v", n.Bounded(), bounded), s)
 		}
 
 		comp = append(comp, n.Type().Components()[1])
@@ -242,9 +239,12 @@ func NewCoGBK(g *Graph, s *Scope, ns []*Node) (*MultiEdge, error) {
 // NewFlatten inserts a new Flatten edge in the graph. Flatten output type is
 // the shared input type.
 func NewFlatten(g *Graph, s *Scope, in []*Node) (*MultiEdge, error) {
+	addContext := func(err error, s *Scope) error {
+		return errors.WithContextf(err, "creating new Flatten in scope %v", s)
+	}
+
 	if len(in) < 2 {
-		err := errors.Errorf("Flatten needs at least 2 input, got %v", len(in))
-		return nil, errors.WithContextf(err, "creating new Flatten in scope %v", s)
+		return nil, addContext(errors.Errorf("Flatten needs at least 2 input, got %v", len(in)), s)
 	}
 	t := in[0].Type()
 	w := inputWindow(in)
@@ -260,17 +260,14 @@ func NewFlatten(g *Graph, s *Scope, in []*Node) (*MultiEdge, error) {
 	}
 	for _, n := range in {
 		if !typex.IsEqual(t, n.Type()) {
-			err := errors.Errorf("mismatched Flatten input types: %v, want %v", n.Type(), t)
-			return nil, errors.WithContextf(err, "creating new Flatten in scope %v", s)
+			return nil, addContext(errors.Errorf("mismatched Flatten input types: %v, want %v", n.Type(), t), s)
 		}
 		if !w.Equals(n.WindowingStrategy()) {
-			err := errors.Errorf("mismatched Flatten window types: %v, want %v", n.WindowingStrategy(), w)
-			return nil, errors.WithContextf(err, "creating new Flatten in scope %v", s)
+			return nil, addContext(errors.Errorf("mismatched Flatten window types: %v, want %v", n.WindowingStrategy(), w), s)
 		}
 	}
 	if typex.IsCoGBK(t) {
-		err := errors.Errorf("Flatten input type cannot be CoGBK: %v", t)
-		return nil, errors.WithContextf(err, "creating new Flatten in scope %v", s)
+		return nil, addContext(errors.Errorf("Flatten input type cannot be CoGBK: %v", t), s)
 	}
 
 	edge := g.NewEdge(s)
@@ -338,14 +335,16 @@ const CombinePerKeyScope = "CombinePerKey"
 // NewCombine inserts a new Combine edge into the graph. Combines cannot have side
 // input.
 func NewCombine(g *Graph, s *Scope, u *CombineFn, in *Node, ac *coder.Coder) (*MultiEdge, error) {
+	addContext := func(err error, s *Scope) error {
+		return errors.WithContextf(err, "creating new Combine in scope %v", s)
+	}
+
 	inT := in.Type()
 	if !typex.IsCoGBK(inT) {
-		err := errors.Errorf("Combine requires CoGBK type: %v", inT)
-		return nil, errors.WithContextf(err, "creating new Combine in scope %v", s)
+		return nil, addContext(errors.Errorf("Combine requires CoGBK type: %v", inT), s)
 	}
 	if len(inT.Components()) > 2 {
-		err := errors.Errorf("Combine cannot follow multi-input CoGBK: %v", inT)
-		return nil, errors.WithContextf(err, "creating new Combine in scope %v", s)
+		return nil, addContext(errors.Errorf("Combine cannot follow multi-input CoGBK: %v", inT), s)
 	}
 
 	// Create a synthetic function for binding purposes. It takes main input
@@ -394,7 +393,7 @@ func NewCombine(g *Graph, s *Scope, u *CombineFn, in *Node, ac *coder.Coder) (*M
 
 	inbound, kinds, outbound, out, err := Bind(synth, nil, inT)
 	if err != nil {
-		return nil, errors.WithContextf(err, "creating new Combine in scope %v", s)
+		return nil, addContext(err, s)
 	}
 
 	edge := g.NewEdge(s)