Only the first file Pair is processed from Channel.fromFilePairs()

Question

This is my first nextflow pipeline and i am struggling with the .fromFilePairs concept of Nextflow.

I want to process paired-end read files and align them against a reference genome. I use Hisat2 so the index files come in form of 8 files with the suffix .ht2. Hisat needs the file path and the filename without ".ht2". All worked well with a single-end version of the process. The reference also has to be passed through a channel because i use a hisat docker-container and otherwise the files are not mounted.

The idea is to pass read-pairs to a preprocessing step and then pass them (as tuples, like a Channel.fromPairs()) into the hisat process and then further downstream.

My problem is all file-pairs are processed during the first step but only the first pair is passed further downstream. All downstream steps finish with the one pair and the rest simply stays in the output-folder of the preprocessing step.

I tried to simply skip the first step to see if the "connection" between the first and second process is not working but the Channel which works for the first step does not for the hisat step. So i think there is a problem with the combination of multiple read-pairs as input but only one reference file?!

process hisat2_paired {
  tag "$sample_id"
 
  container "${params.container_hisat}"
  publishDir "${params.out_dir}/Nextflow_output/hisat/"

  input:
  tuple val(sample_id), path(input_files)
  path( index_files )

  output:
  path('*.sam') , emit: hisat2_sam_out_ch
  path('*.txt')

  script:
  """
  INDEX=`find -L ${index_files}/ -name "*.1.ht2" | sed 's/.1.ht2//'` 
  hisat2 --rna-strandness RF \
          -x \\\$INDEX \
          -1 ${input_files[0]} \
      -2 ${input_files[1]} \
          -S ${sample_id}.sam &> ${sample_id}.hisat_summary.txt
  """

workflow {

Channel
    .fromFilePairs( "${params.reads_folder}/*_{R1,R2}*f*q.gz")
    .ifEmpty { exit 1, "no fastq files found at given path" }
    .view()
    .set{ read_pair_ch }

Channel
    .fromPath( params.hisat2_index )
    .ifEmpty { exit 1, "no hisat2 index files found - path was empty" }
    //.view()
    .set { hisat2_index_ch }

hisat2_paired( read_pair_ch, hisat2_index_ch )

}

I started the workflow with 3 test file-pairs and the Output on the CLI states that only one process is started although if i look at the Channel.view() all 3 file pairs are printed.

executor >  local (1)
[2f/30a4d6] process > hisat2_paired (MC3T3-Ko1_S2_L001) [  0%] 0 of 1
[MC3T3-Ko1_S2_L001, [/mnt/c/Path/to/MC3T3-Ko1_S2_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko1_S2_L001_R2_001.fastq.gz]]
[MC3T3-Ko2_S4_L001, [/mnt/c/Path/to/MC3T3-Ko2_S4_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko2_S4_L001_R2_001.fastq.gz]]
[MC3T3-Ko3_S6_L001, [/mnt/c/Path/to/MC3T3-Ko3_S6_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko3_S6_L001_R2_001.fastq.gz]]

I was already thinking about memory and resources. Since I prototype on my local PC i set the queue size in the config file to 2. I know that my hisat process needs about 7Gb RAM (for Mus.musculus) per process so there should be enough resources. Also the problem was that the local executor is not even trying to start a second process.

Answer 1

So i think there is a problem with the combination of multiple read-pairs as input but only one reference file?!

Correct. The problem is that the hisat2_index_ch is a queue channel, but what you want is a value channel so that the (file) values can be read an unlimited number of times without being consumed. Assuming params.hisat2_index is intended to supply a glob pattern, you could simply call the collect operator to produce a list of files as the sole emission:

Channel
    .fromPath( params.hisat2_index )
    .collect()
    .set { hisat2_index_ch }

You would then need to update your process definition accordingly. Alternatively, have your workflow create the HISAT2 index files for you. This simplifies things considerably, by requiring only a single reference FASTA file as input. Note that value channels are implicitly created as output for a process whose inputs are all value channels. This means in the example below, hisat2_index.out is also a value channel:

params.publish_dir = './results'

params.container_hisat = 'quay.io/biocontainers/hisat2:2.2.1--h87f3376_4'

params.reads = '*_R{1,2}.fastq.gz'
params.ref_fasta = 'chr1.fasta'

process hisat2_index {

    container params.container_hisat
    cpus 8

    input:
    path fasta

    output:
    tuple val(fasta.baseName), path("${fasta.baseName}.*.ht2")

    """
    hisat2-build \\
        -p ${task.cpus} \\
        "${fasta}" \\
        "${fasta.baseName}"
    """
}

process hisat2_paired {

    tag { sample_id }

    container params.container_hisat
    publishDir "${params.publish_dir}/hisat"

    input:
    tuple val(sample_id), path(input_files)
    tuple val(idx), path('hisat2-idx/*')

    output:
    path('*.sam'), emit: hisat2_sam
    path('*.txt')

    script:
    def (fq1, fq2) = input_files

    """
    hisat2 \\
      --rna-strandness RF \\
      -x "hisat2-idx/${idx}" \\
      -1 ${fq1} \\
      -2 ${fq2} \\
      -S ${sample_id}.sam \\
      &> ${sample_id}.hisat_summary.txt
    """
}

workflow {

    read_pairs = Channel.fromFilePairs( params.reads )
    ref_fasta = file( params.ref_fasta )

    hisat2_index( ref_fasta )

    hisat2_paired( read_pairs, hisat2_index.out )
}