[
https://issues.apache.org/jira/browse/BEAM-6825?focusedWorklogId=224689&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-224689
]
ASF GitHub Bot logged work on BEAM-6825:
----------------------------------------
Author: ASF GitHub Bot
Created on: 09/Apr/19 00:03
Start Date: 09/Apr/19 00:03
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #8243: [BEAM-6825]
Improve Combine error messages.
URL: https://github.com/apache/beam/pull/8243#discussion_r273276453
##########
File path: sdks/go/pkg/beam/core/graph/fn.go
##########
@@ -281,35 +282,144 @@ func NewCombineFn(fn interface{}) (*CombineFn, error) {
// AsCombineFn converts a Fn to a CombineFn, if possible.
func AsCombineFn(fn *Fn) (*CombineFn, error) {
+ const fnKind = "graph.AsCombineFn"
if fn.methods == nil {
fn.methods = make(map[string]*funcx.Fn)
}
if fn.Fn != nil {
fn.methods[mergeAccumulatorsName] = fn.Fn
}
- if err := verifyValidNames(fn, setupName, createAccumulatorName,
addInputName, mergeAccumulatorsName, extractOutputName, compactName,
teardownName); err != nil {
+ if err := verifyValidNames(fnKind, fn, setupName,
createAccumulatorName, addInputName, mergeAccumulatorsName, extractOutputName,
compactName, teardownName); err != nil {
return nil, err
}
- if _, ok := fn.methods[mergeAccumulatorsName]; !ok {
- return nil, fmt.Errorf("failed to find %v method: %v",
mergeAccumulatorsName, fn)
+ mergeFn, ok := fn.methods[mergeAccumulatorsName]
+ if !ok {
+ return nil, fmt.Errorf("%v: failed to find required %v method
on type: %v", fnKind, mergeAccumulatorsName, fn.Name())
}
- // TODO(herohde) 5/24/2017: validate the signatures, incl. consistency.
+ // CombineFn methods must satisfy the following:
+ // CreateAccumulator func() (A, error?)
+ // AddInput func(A, I) (A, error?)
+ // MergeAccumulators func(A, A) (A, error?)
+ // ExtractOutput func(A) (O, error?)
+ // This means that the other signatures *must* match the type used in
MergeAccumulators.
+ if len(mergeFn.Ret) <= 0 {
+ return nil, fmt.Errorf("%v: %v requires at least 1 return
value. : %v", fnKind, mergeAccumulatorsName, mergeFn)
+ }
+ accumType := mergeFn.Ret[0].T
+
+ mSig := funcx.Replace(mergeAccumulatorsSig, typex.TType, accumType)
+ if err := funcx.Satisfy(mergeFn, mSig); err != nil {
+ return nil, &verifyMethodError{fnKind, mergeAccumulatorsName,
err, fn, accumType, mSig}
+ }
+
+ if fx, ok := fn.methods[createAccumulatorName]; ok {
+ caSig := funcx.Replace(createAccumulatorSig, typex.TType,
accumType)
+ if err := funcx.Satisfy(fx, caSig); err != nil {
+ return nil, &verifyMethodError{fnKind,
createAccumulatorName, err, fn, accumType, caSig}
+ }
+ }
+ if fx, ok := fn.methods[addInputName]; ok {
+ // AddInput needs the last parameter type substituted.
+ p := fx.Param[len(fx.Param)-1]
+ aiSig := funcx.Replace(addInputSig, typex.TType, accumType)
+ aiSig = funcx.Replace(aiSig, typex.VType, p.T)
+ if err := funcx.Satisfy(fx, aiSig); err != nil {
+ return nil, &verifyMethodError{fnKind, addInputName,
err, fn, accumType, aiSig}
+ }
+ }
+ if fx, ok := fn.methods[extractOutputName]; ok {
+ // ExtractOutput needs the first Return type substituted.
+ r := fx.Ret[0]
+ eoSig := funcx.Replace(extractOutputSig, typex.TType, accumType)
+ eoSig = funcx.Replace(eoSig, typex.WType, r.T)
+ if err := funcx.Satisfy(fx, eoSig); err != nil {
+ return nil, &verifyMethodError{fnKind,
extractOutputName, err, fn, accumType, eoSig}
+ }
+ }
return (*CombineFn)(fn), nil
}
-func verifyValidNames(fn *Fn, names ...string) error {
+func verifyValidNames(fnKind string, fn *Fn, names ...string) error {
m := make(map[string]bool)
for _, name := range names {
m[name] = true
}
for key := range fn.methods {
if !m[key] {
- return fmt.Errorf("unexpected method %v present. Valid
methods are: %v", key, names)
+ return fmt.Errorf("%s: unexpected exported method %v
present. Valid methods are: %v", fnKind, key, names)
}
}
return nil
}
+
+type verifyMethodError struct {
+ // Context for the error.
+ fnKind, methodName string
+ // The triggering error.
+ err error
+
+ fn *Fn
+ accumType reflect.Type
+ sig *funcx.Signature
+}
+
+func (e *verifyMethodError) Error() string {
+ name := e.fn.methods[e.methodName].Fn.Name()
+ if e.fn.Fn == nil {
+ // Methods might be hidden behind reflect.methodValueCall,
which is
+ // not useful to the end user.
+ name = fmt.Sprintf("%s.%s", e.fn.Name(), e.methodName)
+ }
+ typ := e.fn.methods[e.methodName].Fn.Type()
+ switch e.methodName {
+ case mergeAccumulatorsName:
+ // Provide a clearer error for MergeAccumulators, since it's
the root method
+ // for CombineFns.
+ // The root error doesn't matter here since we can't be certain
what the accumulator
+ // type is before mergeAccumulators is verified.
+ return fmt.Sprintf("%v: %s must be a binary merge of
accumulators to be a CombineFn. "+
+ "It is of type \"%v\", but it must be of type
func(context.Context?, A, A) (A, error?) "+
+ "where A is the accumulator type",
+ e.fnKind, name, typ)
+ case createAccumulatorName, addInputName, extractOutputName:
+ // Commonly the accumulator type won't match.
+ if err, ok := e.err.(*funcx.TypeMismatchError); ok && err.Want
== e.accumType {
+ return fmt.Sprintf("%s invalid %v: %s has type \"%v\",
but expected \"%v\" "+
+ "to be the accumulator type \"%v\"; expected a
signature like %v",
+ e.fnKind, e.methodName, name, typ, err.Got,
e.accumType, e.sig)
+ }
+ }
+ return fmt.Sprintf("%s invalid %v %v: \"%v\" for %v; expected a
signature like %v",
Review comment:
This error is much harder for me to parse than the other ones in this
method. I think it could be improved by a few more descriptive words and
structuring it more like other errors in this codebase, where the original
error is appended to the end. Here's an example I think works better:
{e.fnKind} invalid {e.methodName} {name}: got type {typ} but expected a
signature like {e.sig}, original error: {e.err}
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 224689)
Time Spent: 0.5h (was: 20m)
> Improve pipeline construction time error messages in Go SDK.
> ------------------------------------------------------------
>
> Key: BEAM-6825
> URL: https://issues.apache.org/jira/browse/BEAM-6825
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
> Reporter: Daniel Oliveira
> Assignee: Daniel Oliveira
> Priority: Minor
> Labels: triaged
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> Many error messages for common pipeline construction mistakes are unclear and
> unhelpful. They need to be improved to provide more context, especially for
> newer users. This bug tracks these error message improvements.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
