Support process named arguments in typed workflows

Signed-off-by: Ben Sherman <bentshermann@gmail.com>

Author: bentsherman

docs/workflow-typed.md

@@ -181,6 +181,70 @@ The `Value` type supports the following operators:
 
 See {ref}`operator-typed-page` for more information about each operator. See {ref}`migrating-typed-operators` to learn how to migrate existing operators to typed workflows.
 
+### Process named arguments
+
+A common pattern is for a workflow to combine a channel with one or more dataflow values into a single input for a process:
+
+```nextflow
+nextflow.preview.types = true
+
+process align {
+    input:
+    input: Record {
+        id: String
+        fastq: Path
+        index: Path
+    }
+
+    // ...
+}
+
+workflow align_dedup {
+    take:
+    ch_samples: Channel<Sample>
+    index: Value<Path>
+
+    main:
+    ch_align_inputs = ch_samples
+        .cross(index)
+        .map { sample, index -> sample + record(index: index) }
+
+    align( ch_align_inputs )
+
+    // ...
+}
+
+record Sample {
+    id: String
+    fastq: Path
+}
+```
+
+This pattern requires a `cross` and `map` operation for each dataflow value that must be added, which quickly becomes verbose.
+
+In a typed workflow, the same behavior can be achieved by calling the process with named arguments:
+
+```nextflow
+workflow align_dedup {
+    take:
+    ch_samples: Channel<Sample>
+    index: Value<Path>
+
+    main:
+    align(ch_samples, index: index)
+
+    // ...
+}
+```
+
+The named arguments supplied to `align` are automatically added to each record in `ch_samples`, producing the equivalent of `ch_align_inputs` in the previous example.
+
+Named arguments can be used with a process under the following conditions:
+
+- The process declares a single record input  
+- The positional argument (i.e. `ch_samples`) is a channel of records  
+- The named arguments are regular values or dataflow values, not channels
+
 ### Restricted syntax
 
 The following syntax patterns are not supported in typed workflows.

modules/nextflow/src/main/groovy/nextflow/script/ProcessDef.groovy

@@ -33,6 +33,9 @@ import nextflow.script.dsl.ProcessConfigBuilder
 import nextflow.script.params.BaseInParam
 import nextflow.script.params.BaseOutParam
 import nextflow.script.params.EachInParam
+import nextflow.script.params.v2.ProcessInputsDef
+import nextflow.script.types.Record
+import nextflow.util.RecordMap
 
 /**
  * Models a nextflow process definition
@@ -209,6 +212,9 @@ class ProcessDef extends BindableDef implements IterableDef, ChainableDef {
         final declaredInputs = config.getInputs()
         final declaredOutputs = config.getOutputs()
 
+        // combine named args if applicable
+        args = combineNamedArgs(args, declaredInputs)
+
         // validate arguments
         if( args.size() != declaredInputs.size() )
             throw new ScriptRuntimeException(missMatchErrMessage(processName, declaredInputs.size(), args.size()))
@@ -243,6 +249,23 @@ class ProcessDef extends BindableDef implements IterableDef, ChainableDef {
             null
     }
 
+    private Object[] combineNamedArgs(Object[] args, ProcessInputsDef declaredInputs) {
+        if( !(args.length == 2 && args[0] instanceof Map) )
+            return args
+        if( !(declaredInputs.size() == 1 && Record.class.isAssignableFrom(declaredInputs[0].getType())) )
+            return args
+        final opts = (Map<String,Object>) args[0]
+        def result = args[1]
+        for( final name : opts.keySet() ) {
+            final value = opts[name]
+            if( result instanceof ChannelImpl )
+                result = result.cross(value).map { List rv -> ((RecordMap) rv[0]).plus(new RecordMap([(name): rv[1]])) }
+            else if( result instanceof ValueImpl )
+                result = result.cross(value).map { List rv -> ((RecordMap) rv[0]).plus(new RecordMap([(name): rv[1]])) }
+        }
+        return new Object[] { result }
+    }
+
     private DataflowReadChannel createSourceChannel(Object value) {
         if( value instanceof ChannelImpl )
             return CH.getReadChannel(value.getSource())

modules/nf-lang/src/main/java/nextflow/script/control/TypeCheckingVisitor.java

@@ -17,12 +17,18 @@
 
 import nextflow.script.ast.ASTNodeMarker;
 import nextflow.script.ast.ProcessNode;
+import nextflow.script.ast.ProcessNodeV2;
+import nextflow.script.ast.RecordNode;
 import nextflow.script.ast.ScriptNode;
 import nextflow.script.ast.ScriptVisitorSupport;
 import nextflow.script.ast.WorkflowNode;
+import nextflow.script.types.Record;
 import org.codehaus.groovy.ast.ASTNode;
+import org.codehaus.groovy.ast.ClassHelper;
+import org.codehaus.groovy.ast.ClassNode;
 import org.codehaus.groovy.ast.MethodNode;
 import org.codehaus.groovy.ast.expr.MethodCallExpression;
+import org.codehaus.groovy.ast.expr.NamedArgumentListExpression;
 import org.codehaus.groovy.control.SourceUnit;
 import org.codehaus.groovy.control.messages.SyntaxErrorMessage;
 import org.codehaus.groovy.syntax.SyntaxException;
@@ -36,6 +42,8 @@
  */
 public class TypeCheckingVisitor extends ScriptVisitorSupport {
 
+    private static final ClassNode RECORD_TYPE = ClassHelper.makeCached(Record.class);
+
     private SourceUnit sourceUnit;
 
     public TypeCheckingVisitor(SourceUnit sourceUnit) {
@@ -66,8 +74,25 @@ public void visitMethodCallExpression(MethodCallExpression node) {
     private void checkMethodCallArguments(MethodCallExpression node, MethodNode defNode) {
         var argsCount = asMethodCallArguments(node).size();
         var paramsCount = defNode.getParameters().length;
-        if( argsCount != paramsCount )
-            addError(String.format("Incorrect number of call arguments, expected %d but received %d", paramsCount, argsCount), node);
+        if( argsCount == paramsCount )
+            return;
+        if( isProcessCallWithNamedArgs(node) )
+            return;
+        addError(String.format("Incorrect number of call arguments, expected %d but received %d", paramsCount, argsCount), node);
+    }
+
+    private static boolean isProcessCallWithNamedArgs(MethodCallExpression node) {
+        var defNode = (MethodNode) node.getNodeMetaData(ASTNodeMarker.METHOD_TARGET);
+        var args = asMethodCallArguments(node);
+        return defNode instanceof ProcessNodeV2
+            && defNode.getParameters().length == 1
+            && isRecordType(defNode.getParameters()[0].getType())
+            && args.size() == 2
+            && args.get(0) instanceof NamedArgumentListExpression;
+    }
+
+    private static boolean isRecordType(ClassNode type) {
+        return RECORD_TYPE.equals(type) || type.redirect() instanceof RecordNode;
     }
 
     @Override

modules/nf-lang/src/test/groovy/nextflow/script/control/TypeCheckingTest.groovy

@@ -85,4 +85,34 @@ class TypeCheckingTest extends Specification {
         errors[0].getOriginalMessage() == 'Incorrect number of call arguments, expected 2 but received 1'
     }
 
+    def 'should allow typed process to be called with named args' () {
+        when:
+        def errors = check(
+            '''\
+            nextflow.preview.types = true
+
+            process align {
+                input:
+                in: Record {
+                    id: String
+                    fastq: Path
+                    index: Path
+                }
+
+                script:
+                """
+                """
+            }
+
+            workflow {
+                ch_samples = channel.of( record(id: '1', fastq: file('1.fastq')) )
+                index = file('index.fasta')
+                align( ch_samples, index: index )
+            }
+            '''
+        )
+        then:
+        errors.size() == 0
+    }
+
 }