NextFlow: How to fail if channel is empty ( .ifEmpty() )

user_1250832 0
Viewed 331

I'd like for my NextFlow pipeline to fail if a specific channel is empty because, as is, the pipeline will continue as though nothing is wrong, but the process depending on the channel never starts. The answer to a related post states that we generally shouldn't check if a channel is empty, but I'm not sure how else to handle this.

The issue I'm having in the below example is that it always fails, but the process is called if I comment out the .ifEmpty() statement.

Here's a basic example:

/*
 * There are .cram files in this folder
 */
params.input_sample_folder = 'path/to/folder/*' 
samples = Channel.fromPath(params.input_sample_folder, checkIfExists: true)
                 .filter( ~/.*(\.sam|\.bam|\.cram)/ )
                 .ifEmpty( exit 1,
                  "ERROR: Did not find any samples in ${params.input_sample_folder})

workflow{
    PROCESS_SAMPLES( samples )
}

Ultimate questions:

  1. My guess is that the channel does not fill immediately. Is that true? If so, when does it fill?
  2. How should I handle this situation? I want to fail if the channel doesn't get populated. e.g., I was surprised to learn that the channel remains empty if I only provide a folder path without a glob/wildcard character (/path/to/folder/; no * or *.cram, etc.). I don't think I can handle it in the process itself, because the process never gets called if the channel is legitimately empty.

Really appreciate your help.

nextflow
1 Answers

Setting checkIfExists: true will actually throw an exception for you if the specified files do not exist on your file system. The trick is to specify the files you need when you create the channel, rather than filtering for them downstream. For example, all you need is:

params.input_sample_folder = 'path/to/folder'

samples = Channel.fromPath(
    "${params.input_sample_folder}/*.{sam,bam,cram}",
    checkIfExists: true,
)

Or arguably better; since this gives the user full control over the input files:

params.input_sample_files = 'path/to/folder/*.{sam,bam,cram}'

samples = Channel.fromPath( params.input_sample_files, checkIfExists: true )

Either way, both will have your pipeline fail with exit status 1 and the following message in red when no matching files exist:

No files match pattern `*.{sam,bam,cram}` at path: path/to/folder/

As per the docs, the ifEmpty operator is really just intended to emit a default value when a channel becomes empty. To avoid having to check if a channel is empty, the general solution is to just avoid creating an empty channel in the first place. There's lots of ways to do this, but one way might look like:

import org.apache.log4j.Logger

nextflow.enable.dsl=2


def find_sample_files( input_dir ) {

    def pattern = ~/.*(\.sam|\.bam|\.cram)/
    def results = []

    input_dir.eachFileMatch(pattern) { item ->
        results.add( item )
    }

    return results
}


params.input_sample_folder = 'path/to/folder'

workflow {

    input_sample_folder = file( params.input_sample_folder )
    input_sample_files = find_sample_files( input_sample_folder )

    if ( !input_sample_files ) {
        log.error("ERROR: Did not find any samples in ${params.input_sample_folder}")
        System.exit(1)
    }

    sample_files = Channel.of( input_sample_files )
    sample_files.view()
}
user_7518630
Related