Add arity option for path inputs and outputs

docs/process.rst

@@ -587,10 +587,10 @@ will output::
 
 The target input file name may contain the ``*`` and ``?`` wildcards, which can be used
 to control the name of staged files. The following table shows how the wildcards are
-replaced depending on the cardinality of the received input collection.
+replaced depending on the arity of the received input collection.
 
 ============ ============== ==================================================
-Cardinality   Name pattern   Staged file names
+Arity        Name pattern   Staged file names
 ============ ============== ==================================================
  any         ``*``           named as the source file
  1           ``file*.ext``   ``file.ext``
@@ -655,6 +655,26 @@ with the current execution context.
   and some of these files may have the same filename. In this case, a solution would be to use
   the option ``stageAs``.
 
+
+Input file arity
+----------------
+
+.. note::
+  This feature requires Nextflow version 23.04.0 or later.
+
+The *arity* of a ``path`` input is the number of files that it is expected to contain. The arity can be a
+number or a range. Here are some examples::
+
+    input:
+        path('one.txt', arity: '1')         // exactly one file is expected
+        path('pair_*.txt', arity: '2')      // exactly two files are expected
+        path('many_*.txt', arity: '1..*')   // one or more files are expected
+        path('optional.txt', arity: '0..1') // zero or one file is expected
+
+When a task is created, Nextflow will check whether the received files for each path input match the
+declared arity, and fail if they do not. The default arity is ``'1..*'``.
+
+
 Input type ``env``
 ------------------
 
@@ -1150,6 +1170,28 @@ on the actual value of the ``species`` input.
   because it will result in simpler and more portable code.
 
 
+Output file arity
+-----------------
+
+.. note::
+  This feature requires Nextflow version 23.04.0 or later.
+
+The *arity* of a ``path`` output is the number of files that it is expected to contain. The arity can be a
+number or a range. Here are some examples::
+
+    output:
+        path('one.txt', arity: '1')         // exactly one file is expected
+        path('pair_*.txt', arity: '2')      // exactly two files are expected
+        path('many_*.txt', arity: '1..*')   // one or more files are expected
+        path('optional.txt', arity: '0..1') // zero or one file is expected (equivalent to optional: true)
+
+When a task completes, Nextflow will check whether the produced files for each path output match the
+declared arity, and fail if they do not.
+
+The default arity is ``'1..*'``, or ``'0..*'`` if the output is declared optional. If you declare an
+output as optional and also give it an arity, the optional flag will be ignored.
+
+
 .. _process-env:
 
 Output type ``env``

modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy

@@ -1570,15 +1570,18 @@ class TaskProcessor {
             if( result )
                 allFiles.addAll(result)
 
-            else if( !param.optional ) {
+            else if( !param.optional && (!param.arity || param.arity.min > 0) ) {
                 def msg = "Missing output file(s) `$filePattern` expected by process `${safeTaskName(task)}`"
                 if( inputsRemovedFlag )
                     msg += " (note: input files are not included in the default matching set)"
                 throw new MissingFileException(msg)
             }
         }
 
-        task.setOutput( param, allFiles.size()==1 ? allFiles[0] : allFiles )
+        if( param.arity && !param.arity.contains(allFiles.size()) )
+            throw new IllegalArgumentException("Incorrect number of output files for process `${safeTaskName(task)}` -- expected ${param.arity}, found ${allFiles.size()}")
+
+        task.setOutput( param, allFiles.size()==1 && (!param.arity || param.arity.isSingle()) ? allFiles[0] : allFiles )
 
     }
 
@@ -1821,10 +1824,10 @@ class TaskProcessor {
         return files
     }
 
-    protected singleItemOrList( List<FileHolder> items, ScriptType type ) {
+    protected singleItemOrList( List<FileHolder> items, boolean single, ScriptType type ) {
         assert items != null
 
-        if( items.size() == 1 ) {
+        if( items.size() == 1 && single ) {
             return makePath(items[0],type)
         }
 
@@ -2022,17 +2025,22 @@ class TaskProcessor {
             final param = entry.getKey()
             final val = entry.getValue()
             final fileParam = param as FileInParam
-            final normalized = normalizeInputToFiles(val, count, fileParam.isPathQualifier(), batch)
-            final resolved = expandWildcards( fileParam.getFilePattern(ctx), normalized )
-            ctx.put( param.name, singleItemOrList(resolved, task.type) )
-            count += resolved.size()
-            for( FileHolder item : resolved ) {
+
+            def files = normalizeInputToFiles(val, count, fileParam.isPathQualifier(), batch)
+            files = expandWildcards( fileParam.getFilePattern(ctx), files )
+
+            if( param.arity && !param.arity.contains(files.size()) )
+                throw new IllegalArgumentException("Incorrect number of input files for process `${safeTaskName(task)}` -- expected ${param.arity}, found ${files.size()}")
+
+            ctx.put( param.name, singleItemOrList(files, !param.arity || param.arity.isSingle(), task.type) )
+            count += files.size()
+            for( FileHolder item : files ) {
                 Integer num = allNames.getOrCreate(item.stageName, 0) +1
                 allNames.put(item.stageName,num)
             }
 
             // add the value to the task instance context
-            task.setInput(param, resolved)
+            task.setInput(param, files)
         }
 
         // -- set the delegate map as context in the task config

modules/nextflow/src/main/groovy/nextflow/script/params/ArityParam.groovy

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2023, Seqera Labs
+ *
+ * Licensed 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 nextflow.script.params
+
+import groovy.transform.CompileStatic
+
+/**
+ * Implements an arity option for process inputs and outputs.
+ *
+ * @author Ben Sherman <[email protected]>
+ */
+@CompileStatic
+trait ArityParam {
+
+    Range arity
+
+    Range getArity() { arity }
+
+    def setArity(String value) {
+        if( value.isInteger() ) {
+            def n = value.toInteger()
+            this.arity = new Range(n, n)
+            return this
+        }
+
+        def tokens = value.tokenize('..')
+        if( tokens.size() == 2 ) {
+            def min = tokens[0]
+            def max = tokens[1]
+            if( min.isInteger() && (max == '*' || max.isInteger()) ) {
+                this.arity = new Range(
+                    min.toInteger(),
+                    max == '*' ? Integer.MAX_VALUE : max.toInteger()
+                )
+                return this
+            }
+        }
+
+        throw new IllegalArgumentException("Path arity should be a number (e.g. '1') or a range (e.g. '1..*')")
+    }
+
+    static class Range {
+        int min
+        int max
+
+        Range(int min, int max) {
+            this.min = min
+            this.max = max
+        }
+
+        boolean contains(int value) {
+            min <= value && value <= max
+        }
+
+        boolean isSingle() {
+            max == 1
+        }
+
+        @Override
+        boolean equals(Object obj) {
+            min == obj.min && max == obj.max
+        }
+
+        @Override
+        String toString() {
+            min == max
+                ? min.toString()
+                : "${min}..${max == Integer.MAX_VALUE ? '*' : max}".toString()
+        }
+    }
+
+}

modules/nextflow/src/main/groovy/nextflow/script/params/FileInParam.groovy

@@ -28,7 +28,7 @@ import nextflow.script.TokenVar
  */
 @Slf4j
 @InheritConstructors
-class FileInParam extends BaseInParam implements PathQualifier {
+class FileInParam extends BaseInParam implements ArityParam, PathQualifier {
 
     protected filePattern
 

modules/nextflow/src/main/groovy/nextflow/script/params/FileOutParam.groovy

@@ -33,7 +33,7 @@ import nextflow.util.BlankSeparatedList
  */
 @Slf4j
 @InheritConstructors
-class FileOutParam extends BaseOutParam implements OutParam, OptionalParam, PathQualifier {
+class FileOutParam extends BaseOutParam implements OutParam, ArityParam, OptionalParam, PathQualifier {
 
     /**
      * ONLY FOR TESTING DO NOT USE

modules/nextflow/src/test/groovy/nextflow/processor/TaskProcessorTest.groovy

@@ -185,13 +185,19 @@ class TaskProcessorTest extends Specification {
 
         when:
         def list = [ FileHolder.get(path1, 'x_file_1') ]
-        def result = processor.singleItemOrList(list, ScriptType.SCRIPTLET)
+        def result = processor.singleItemOrList(list, true, ScriptType.SCRIPTLET)
         then:
         result.toString() == 'x_file_1'
 
+        when:
+        list = [ FileHolder.get(path1, 'x_file_1') ]
+        result = processor.singleItemOrList(list, false, ScriptType.SCRIPTLET)
+        then:
+        result*.toString() == ['x_file_1']
+
         when:
         list = [ FileHolder.get(path1, 'x_file_1'), FileHolder.get(path2, 'x_file_2'), FileHolder.get(path3, 'x_file_3') ]
-        result = processor.singleItemOrList(list, ScriptType.SCRIPTLET)
+        result = processor.singleItemOrList(list, false, ScriptType.SCRIPTLET)
         then:
         result*.toString() == [ 'x_file_1',  'x_file_2',  'x_file_3']
 

modules/nextflow/src/test/groovy/nextflow/script/params/ArityParamTest.groovy

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2023, Seqera Labs
+ *
+ * Licensed 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 nextflow.script.params
+
+import spock.lang.Specification
+import spock.lang.Unroll
+/**
+ *
+ * @author Ben Sherman <[email protected]>
+ */
+class ArityParamTest extends Specification {
+
+    static class DefaultArityParam implements ArityParam {
+        DefaultArityParam() {}
+    }
+
+    @Unroll
+    def testArity () {
+
+        when:
+        def param = new DefaultArityParam()
+        param.setArity(VALUE)
+        then:
+        param.arity.min == MIN
+        param.arity.max == MAX
+
+        where:
+        VALUE  | MIN | MAX
+        '1'    | 1   | 1
+        '0..1' | 0   | 1
+        '1..*' | 1   | Integer.MAX_VALUE
+    }
+
+    @Unroll
+    def testArityRange () {
+
+        when:
+        def range = new ArityParam.Range(MIN, MAX)
+        then:
+        range.contains(2) == TWO
+        range.isSingle() == SINGLE
+        range.toString() == STRING
+
+        where:
+        MIN | MAX               | TWO   | SINGLE | STRING
+        1   | 1                 | false | true   | '1'
+        0   | 1                 | false | true   | '0..1'
+        1   | Integer.MAX_VALUE | true  | false  | '1..*'
+    }
+
+}

modules/nextflow/src/test/groovy/nextflow/script/params/ParamsInTest.groovy

@@ -706,12 +706,12 @@ class ParamsInTest extends Dsl2Spec {
 
             process hola {
               input:
-              path x
-              path f1
-              path '*.fa' 
+              path x, arity: '1'
+              path f1, arity: '1..2'
+              path '*.fa', arity: '1..*'
               path 'file.txt'
               path f2, name: '*.fa'
-              path f3, stageAs: '*.txt' 
+              path f3, stageAs: '*.txt'
 
               return ''
             }
@@ -737,18 +737,21 @@ class ParamsInTest extends Dsl2Spec {
         in0.inChannel.val == FILE
         in0.index == 0
         in0.isPathQualifier()
+        in0.arity == new ArityParam.Range(1, 1)
 
         in1.name == 'f1'
         in1.filePattern == '*'
         in1.inChannel.val == FILE
         in1.index == 1
         in1.isPathQualifier()
+        in1.arity == new ArityParam.Range(1, 2)
 
         in2.name == '*.fa'
         in2.filePattern == '*.fa'
         in2.inChannel.val == FILE
         in2.index == 2
         in2.isPathQualifier()
+        in2.arity == new ArityParam.Range(1, Integer.MAX_VALUE)
 
         in3.name == 'file.txt'
         in3.filePattern == 'file.txt'

modules/nextflow/src/test/groovy/nextflow/script/params/ParamsOutTest.groovy

@@ -931,7 +931,8 @@ class ParamsOutTest extends Dsl2Spec {
                 separatorChar: '#',
                 glob: false,
                 optional: false,
-                includeInputs: false
+                includeInputs: false,
+                arity: '1'
 
               path y, 
                 maxDepth:5,
@@ -941,7 +942,8 @@ class ParamsOutTest extends Dsl2Spec {
                 separatorChar: ':',
                 glob: true,
                 optional: true,
-                includeInputs: true
+                includeInputs: true,
+                arity: '0..*'
 
               return ''
             }
@@ -963,6 +965,7 @@ class ParamsOutTest extends Dsl2Spec {
         !out0.getGlob()
         !out0.getOptional()
         !out0.getIncludeInputs()
+        out0.getArity() == new ArityParam.Range(1, 1)
 
         and:
         out1.getMaxDepth() == 5
@@ -973,6 +976,7 @@ class ParamsOutTest extends Dsl2Spec {
         out1.getGlob()
         out1.getOptional()
         out1.getIncludeInputs()
+        out1.getArity() == new ArityParam.Range(0, Integer.MAX_VALUE)
     }
 
     def 'should set file options' () {