Validate and print the value of Channel and Value params instead of hanging

CHANGELOG.md

@@ -5,6 +5,7 @@
 ## Bug fixes
 
 1. Fixed an issue where a schema referenced with the `schema` keyword was only looked up relative to the project that is running. It is now looked up next to the schema that references it first, and relative to the project otherwise, so a pipeline schema keeps working when it is read from another directory (e.g. when the pipeline is included in another one).
+2. Fixed an issue where `validateParameters()` would hang, and the params summary would print an object name, when a parameter holds a `Channel` or `Value` (e.g. a typed `Channel<...>` param). The value the parameter was created from (from the `params` config scope, which also holds the values given on the command line and in a params file) is validated and printed in its place, also when the dataflow value is nested in a record parameter. A parameter without such a value is left out.
 
 # Version 3.0.0
 

docs/parameters/validation.md

@@ -143,3 +143,9 @@ For example, providing an integer as a string will no longer fail validation.
     It attempts to cast a temporary copy of the params only, during the validation step.
 
 To enable lenient validation mode, set `validation.lenientMode = true` in your configuration file.
+
+## Parameters that are a `Channel` or a `Value`
+
+With typed parameters, a parameter can be declared as a `Channel` or a `Value` (see the [typed parameters](https://github.com/nextflow-io/nextflow/blob/master/docs/typed-parameters.mdx) documentation of Nextflow). For example, a `Channel<Sample>` parameter takes the path of a samplesheet, which Nextflow loads as a channel, and a pipeline that is included in another pipeline can be given a channel by the including pipeline instead.
+
+These parameters hold a dataflow object while the pipeline runs, so there is no value to validate in them. `validateParameters()` validates the value that the parameter was created from instead. Nextflow keeps it in the `params` scope of the configuration, which also holds the values given on the command line and in a params file. A parameter that has no such value, such as one that is given by the dataflow of an including pipeline, is not validated. The same value is shown by [`paramsSummaryLog()` and `paramsSummaryMap()`](summary_log.md), and a parameter without one is left out of the summary.

src/main/groovy/nextflow/validation/ValidationExtension.groovy

@@ -3,6 +3,7 @@ package nextflow.validation
 import static nextflow.validation.utils.Colors.getLogColors
 import static nextflow.validation.utils.Common.getBasePath
 import static nextflow.validation.utils.Common.getLongestKeyLength
+import static nextflow.validation.utils.Common.replaceDataflowParams
 
 import groovy.json.JsonBuilder
 import groovy.util.logging.Slf4j
@@ -197,7 +198,7 @@ class ValidationExtension extends PluginExtensionPoint {
             options,
             session.workflowMetadata,
             session.baseDir,
-            session.params
+            replaceDataflowParams(session.params, session.config?.params)
         )
     }
 

src/main/groovy/nextflow/validation/parameters/ParameterValidator.groovy

@@ -1,5 +1,6 @@
 package nextflow.validation.parameters
 
+import static nextflow.validation.utils.Common.replaceDataflowParams
 import static nextflow.NF.isSyntaxParserV2
 
 import static nextflow.validation.utils.Colors.getLogColors
@@ -142,17 +143,18 @@ class ParameterValidator {
             .addConverter(MemoryUnit) { MemoryUnit memory -> memory.toBytes() }
             .addConverter(VersionNumber) { VersionNumber version -> version.toString() }
 
-        // Cast parameters provided via the CLI to their respective types.
-        // This is a temporary workaround until static typing is introduced in Nextflow,
-        // in which case we can rely on the static type system to do the casting for us.
-        // This mimics the type casting behaviour of syntax parser V1 so shouldn't introduce any breaking changes.
-        if (castCliParams) {
-            List<String> cliParams = (session.cliParams?.keySet()?.toList()*.toString() ?: []) as List<String>
-            generatorOptions.addConverter(Map<String, Object>) { Map<String,Object> map ->
-                map.collectEntries { k, v ->
-                    // Only cast parameters that were explicitly provided via the CLI
-                    return (cliParams.contains(k) && v in String) ? [k, parseParamValue(v as String)] : [k, v]
-                }
+        // The generator calls this converter for the params map before it reads any entry, so the dataflow
+        // params are replaced once on the root map and the nested maps it reaches afterwards hold none.
+        // Parameters provided via the CLI are cast here too (a temporary workaround until static typing is
+        // introduced in Nextflow), which mimics the type casting behaviour of syntax parser V1.
+        List<String> cliParams = castCliParams ?
+            (session.cliParams?.keySet()?.toList()*.toString() ?: []) as List<String> :
+            []
+        generatorOptions.addConverter(Map) { Map map ->
+            Map<Object, Object> level = map.is(params) ? replaceDataflowParams(map, session.config?.params) : map
+            return level.collectEntries { Object k, Object v ->
+                // Only cast parameters that were explicitly provided via the CLI
+                (cliParams.contains(k) && v in String) ? [k, parseParamValue(v as String)] : [k, v]
             }
         }
 

src/main/groovy/nextflow/validation/utils/Common.groovy

@@ -1,5 +1,9 @@
 package nextflow.validation.utils
 
+import groovyx.gpars.dataflow.DataflowReadChannel
+import groovyx.gpars.dataflow.DataflowWriteChannel
+import nextflow.dataflow.ChannelImpl
+import nextflow.dataflow.ValueImpl
 import org.json.JSONObject
 import org.json.JSONArray
 import org.json.JSONPointer
@@ -136,4 +140,37 @@ public class Common {
         }
     }
 
+    // The values a Channel or Value param holds: the typed wrappers and the dataflow channels and variables
+    // underneath them
+    static boolean isDataflowValue(Object value) {
+        return value in ChannelImpl || value in ValueImpl ||
+            value in DataflowReadChannel || value in DataflowWriteChannel
+    }
+
+    // Channel and Value params hold live dataflow objects: reading them blocks, and they print as object
+    // names. The value they were created from is used in their place, and a param without one is left out.
+    // That value is found in the params scope of the config, which also holds the values given on the
+    // command line and in a params file. Params nested in a record (e.g. the params of an included pipeline)
+    // are handled the same way.
+    static Map replaceDataflowParams(Map params, Object configParams) {
+        return replaceDataflowValues(params, configParams) as Map
+    }
+
+    private static Object replaceDataflowValues(Object value, Object configValue) {
+        if (isDataflowValue(value)) {
+            return configValue != null && !isDataflowValue(configValue) ? configValue : null
+        }
+        if (value in Map) {
+            Map<Object, Object> result = [:]
+            (value as Map<Object, Object>).each { Object name, Object entry ->
+                Object replaced = replaceDataflowValues(entry, configValue in Map ? (configValue as Map)[name] : null)
+                if (replaced != null || !isDataflowValue(entry)) {
+                    result[name] = replaced
+                }
+            }
+            return result
+        }
+        return value
+    }
+
 }

src/test/groovy/nextflow/validation/ParamsSummaryLogTest.groovy

@@ -14,6 +14,7 @@ import nextflow.plugin.extension.PluginExtensionProvider
 import org.junit.Rule
 import org.pf4j.PluginDescriptorFinder
 import spock.lang.Shared
+import spock.lang.Timeout
 import test.Dsl2Spec
 import test.OutputCapture
 
@@ -112,6 +113,58 @@ class ParamsSummaryLogTest extends Dsl2Spec {
         stdout ==~ /.*outdir     : outDir.*/
     }
 
+    @Timeout(60)
+    void 'should leave a param that holds a dataflow value out of the params summary'() {
+        given:
+        String schema = Path.of('src/testResources/nextflow_schema.json').toAbsolutePath()
+        String script = """
+            include { paramsSummaryLog } from 'plugin/nf-schema'
+            workflow {
+                params.outdir = 'outDir'
+                params.input = new groovyx.gpars.dataflow.DataflowVariable()
+                log.info paramsSummaryLog(parameters_schema: '${schema}')
+            }
+        """
+
+        when:
+        Map opts = ['config': ['validation': ['monochromeLogs': true]]]
+        runScript(opts, script)
+        String stdout = capture
+
+        then:
+        noExceptionThrown()
+        stdout ==~ /(?s).*outdir     : outDir.*/
+        !stdout.contains('DataflowVariable')
+        !stdout.contains('input ')
+    }
+
+    @Timeout(60)
+    void 'should print the config value of a param that holds a dataflow value in the params summary'() {
+        given:
+        String schema = Path.of('src/testResources/nextflow_schema.json').toAbsolutePath()
+        String script = """
+            include { paramsSummaryLog } from 'plugin/nf-schema'
+            workflow {
+                params.outdir = 'outDir'
+                params.input = new groovyx.gpars.dataflow.DataflowVariable()
+                log.info paramsSummaryLog(parameters_schema: '${schema}')
+            }
+        """
+
+        when:
+        Map opts = ['config': [
+            'validation': ['monochromeLogs': true],
+            'params': ['input': 'src/testResources/correct.csv']
+        ]]
+        runScript(opts, script)
+        String stdout = capture
+
+        then:
+        noExceptionThrown()
+        stdout ==~ /(?s).*input      : src\/testResources\/correct.csv.*/
+        !stdout.contains('DataflowVariable')
+    }
+
     void 'should print params summary - nested parameters'() {
         given:
         String schema = Path.of('src/testResources/nextflow_schema_nested_parameters.json').toAbsolutePath()

src/test/groovy/nextflow/validation/ValidateDataflowParamsTest.groovy

@@ -0,0 +1,172 @@
+/* groovylint-disable LineLength, MethodName */
+package nextflow.validation
+
+import groovy.transform.CompileDynamic
+import groovyx.gpars.dataflow.DataflowQueue
+import groovyx.gpars.dataflow.DataflowVariable
+import nextflow.Session
+import nextflow.dataflow.ChannelImpl
+import nextflow.dataflow.ValueImpl
+import nextflow.validation.config.ValidationConfig
+import nextflow.validation.exceptions.SchemaValidationException
+import nextflow.validation.parameters.ParameterValidator
+import spock.lang.Specification
+import spock.lang.Timeout
+
+import java.nio.file.Path
+
+/**
+ * Validation of params that hold a dataflow value (a typed `Channel` or `Value` param), where the value
+ * the param was created from is validated in place of the dataflow object. Nextflow keeps that value in
+ * the params scope of the config, which also holds the values given on the command line and in a params
+ * file. The session is mocked so that the config params can be set directly.
+ */
+@CompileDynamic
+@Timeout(60)
+class ValidateDataflowParamsTest extends Specification {
+
+    private static final String SCHEMA = 'src/testResources/nextflow_schema.json'
+    private static final String NESTED_SCHEMA = 'src/testResources/nextflow_schema_nested_parameters.json'
+
+    void 'should accept a valid value of a dataflow param'() {
+        given:
+        Session session = mockSession(topLevelParams(new DataflowVariable()), [input: 'src/testResources/correct.csv'])
+
+        when:
+        validate(session, SCHEMA)
+
+        then:
+        noExceptionThrown()
+    }
+
+    void 'should reject an invalid value of a dataflow param'() {
+        given:
+        Session session = mockSession(topLevelParams(new DataflowVariable()), [input: 'src/testResources/correct.txt'])
+
+        when:
+        validate(session, SCHEMA)
+
+        then:
+        SchemaValidationException error = thrown(SchemaValidationException)
+        error.message.contains('--input (src/testResources/correct.txt)')
+    }
+
+    void 'should accept a valid value of a Value param'() {
+        given:
+        Session session = mockSession(topLevelParams(new ValueImpl(new DataflowVariable())), [input: 'src/testResources/correct.csv'])
+
+        when:
+        validate(session, SCHEMA)
+
+        then:
+        noExceptionThrown()
+    }
+
+    void 'should reject an invalid value of a Value param'() {
+        given:
+        Session session = mockSession(topLevelParams(new ValueImpl(new DataflowVariable())), [input: 'src/testResources/correct.txt'])
+
+        when:
+        validate(session, SCHEMA)
+
+        then:
+        SchemaValidationException error = thrown(SchemaValidationException)
+        error.message.contains('--input (src/testResources/correct.txt)')
+    }
+
+    void 'should accept a valid value of a Channel param'() {
+        given:
+        Session session = mockSession(topLevelParams(new ChannelImpl(new DataflowQueue())), [input: 'src/testResources/correct.csv'])
+
+        when:
+        validate(session, SCHEMA)
+
+        then:
+        noExceptionThrown()
+    }
+
+    void 'should reject an invalid value of a Channel param'() {
+        given:
+        Session session = mockSession(topLevelParams(new ChannelImpl(new DataflowQueue())), [input: 'src/testResources/correct.txt'])
+
+        when:
+        validate(session, SCHEMA)
+
+        then:
+        SchemaValidationException error = thrown(SchemaValidationException)
+        error.message.contains('--input (src/testResources/correct.txt)')
+    }
+
+    void 'should accept a valid value of a dataflow param when the command line values are not cast'() {
+        given:
+        Session session = mockSession(topLevelParams(new ValueImpl(new DataflowVariable())), [input: 'src/testResources/correct.csv'])
+
+        when:
+        validate(session, SCHEMA, [cast_cli_params: false])
+
+        then:
+        noExceptionThrown()
+    }
+
+    void 'should reject an invalid value of a dataflow param when the command line values are not cast'() {
+        given:
+        Session session = mockSession(topLevelParams(new ValueImpl(new DataflowVariable())), [input: 'src/testResources/correct.txt'])
+
+        when:
+        validate(session, SCHEMA, [cast_cli_params: false])
+
+        then:
+        SchemaValidationException error = thrown(SchemaValidationException)
+        error.message.contains('--input (src/testResources/correct.txt)')
+    }
+
+    void 'should accept a valid value of a dataflow value nested in a record param'() {
+        given:
+        Session session = mockSession(
+            [map: [is: [so: [deep: new ValueImpl(new DataflowVariable())]]]],
+            [map: [is: [so: [deep: true]]]]
+        )
+
+        when:
+        validate(session, NESTED_SCHEMA)
+
+        then:
+        noExceptionThrown()
+    }
+
+    void 'should reject an invalid value of a dataflow value nested in a record param'() {
+        given:
+        Session session = mockSession(
+            [map: [is: [so: [deep: new ValueImpl(new DataflowVariable())]]]],
+            [map: [is: [so: [deep: 'maybe']]]]
+        )
+
+        when:
+        validate(session, NESTED_SCHEMA)
+
+        then:
+        SchemaValidationException error = thrown(SchemaValidationException)
+        error.message.contains('--map.is.so.deep')
+    }
+
+    private Session mockSession(Map params, Map configParams) {
+        Session session = Mock(Session)
+        session.params >> params
+        session.config >> [params: configParams]
+        session.baseDir >> Path.of('.').toAbsolutePath()
+        return session
+    }
+
+    private void validate(Session session, String schema, Map options = [:]) {
+        ValidationConfig config = new ValidationConfig([monochromeLogs: true], session)
+        new ParameterValidator(config).validateParametersMap(
+            [parameters_schema: Path.of(schema).toAbsolutePath().toString()] + options,
+            session
+        )
+    }
+
+    private Map topLevelParams(Object input) {
+        return [input: input, outdir: 'src/testResources/testDir']
+    }
+
+}