youngoli commented on a change in pull request #10991: [BEAM-3301] Refactor
DoFn validation & allow specifying main inputs.
URL: https://github.com/apache/beam/pull/10991#discussion_r391958266
##########
File path: sdks/go/pkg/beam/core/graph/fn.go
##########
@@ -239,52 +279,50 @@ func AsDoFn(fn *Fn) (*DoFn, error) {
return nil, addContext(err, fn)
}
- // Start validating DoFn. First, check that ProcessElement has a main
input.
+ // Validate ProcessElement has correct number of main inputs (as
indicated by
+ // numMainIn), and that main inputs are before side inputs.
processFn := fn.methods[processElementName]
- pos, num, ok := processFn.Inputs()
- if ok {
- first := processFn.Param[pos].Kind
- if first != funcx.FnValue {
- err := errors.New("side input parameters must follow
main input parameter")
- err = errors.SetTopLevelMsgf(err,
- "Method %v of DoFns should always have a main
input before side inputs, "+
- "but it has side inputs (as Iters or
ReIters) first in DoFn %v.",
- processElementName, fn.Name())
- err = errors.WithContextf(err, "method %v",
processElementName)
- return nil, addContext(err, fn)
- }
+ if err := validateMainInputs(fn, processFn, processElementName,
numMainIn); err != nil {
+ return nil, addContext(err, fn)
+ }
+
+ // If numMainIn is unknown, we can try inferring it from the second
input in ProcessElement.
+ // If there is none, or it's not a FnValue type, then we can safely
infer that there's only
+ // one main input.
+ pos, num, _ := processFn.Inputs()
+ if numMainIn == MainUnknown && (num == 1 || processFn.Param[pos+1].Kind
!= funcx.FnValue) {
+ numMainIn = MainSingle
}
// If the ProcessElement function includes side inputs or emit
functions those must also be
Review comment:
It's part of the API for start/finishBundle. I don't remember why it's done
that way though. lostluck@ might be able to answer why when he gets back.
There might be room to make the side inputs/emits in start/finishBundle
optional, but I believe right now it's mandatory (if we don't catch and throw
an error here, it'll just break later on in translation or execution or
something).
----------------------------------------------------------------
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]
With regards,
Apache Git Services
