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

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 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 the command line value of a nested parameter (e.g. `--group.flag true`) was not cast to the type in the schema, while the value of a top-level parameter was. The nested values of the command line parameters are now followed to find the values to cast.
+3. 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,6 +143,10 @@ class ParameterValidator {
             .addConverter(MemoryUnit) { MemoryUnit memory -> memory.toBytes() }
             .addConverter(VersionNumber) { VersionNumber version -> version.toString() }
 
+        // A `Channel` or `Value` parameter is replaced by the value it was created from before the CLI values
+        // are cast, so that a value given on the command line is cast to its type in the schema.
+        params = replaceDataflowParams(params, session.config?.params) as Map<String, Object>
+
         // 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.

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,40 @@ 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.
+    // TODO if Nextflow gains a core way to get these plain values
+    // (https://github.com/nextflow-io/nextflow/issues/7758), consider using it here instead,
+    // see https://github.com/nextflow-io/nf-schema/pull/230
+    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,199 @@
+/* groovylint-disable LineLength, MethodName */
+package nextflow.validation
+
+import groovy.transform.CompileDynamic
+import groovyx.gpars.dataflow.DataflowQueue
+import groovyx.gpars.dataflow.DataflowVariable
+import nextflow.Global
+import nextflow.ISession
+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'
+
+    private ISession previousSession
+
+    void setup() {
+        previousSession = Global.session
+    }
+
+    void cleanup() {
+        Global.session = previousSession
+    }
+
+    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')
+    }
+
+    void 'should cast the command line value of a dataflow param'() {
+        given:
+        Map cliParams = [map: [is: [so: [deep: 'true']]]]
+        Session session = mockSession([map: [is: [so: [deep: new ValueImpl(new DataflowVariable())]]]], cliParams, cliParams)
+
+        when:
+        validate(session, NESTED_SCHEMA, [cast_cli_params: true])
+
+        then:
+        noExceptionThrown()
+    }
+
+    private Session mockSession(Map params, Map configParams, Map cliParams = null) {
+        Session session = Mock(Session)
+        session.params >> params
+        session.cliParams >> cliParams
+        session.config >> [params: configParams]
+        session.baseDir >> Path.of('.').toAbsolutePath()
+        // the schema evaluators read the session from Global
+        Global.session = session
+        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']
+    }
+
+}