Why does this nextflow script finish after running one sample?

Question

I have 36 samples in total in the bam_files folder, with the name like this 20230306_CH_EP_C01.md.bam

I expect the code to output all the 36 samples, one by one. But the run stopped after only running one sample. I have searched online, but did not find a solution. I am using nextflow version 2.

bam_chl = Channel
     .fromPath("bam_files/*.md.bam")
     .map{ item -> tuple(item.baseName.substring(0, 18), item) }
     

     
ref_chl = Channel
      .fromPath("sequences/hg38.fa")
ref_index_chl = Channel
      .fromPath("sequences/hg38.fa.fai")
amplicon_chl = Channel
      .fromPath("sequences/tp53_amplicon.hg38.interval_list")
target_chl = Channel
      .fromPath("sequences/TP53.hg38.interval_list")
     


process get_pcr_metric {

  publishDir path: 'pcr_metric', pattern: '*.csv'
  
  input:
  tuple val(pair_id), path(bam)
  path(ref)
  path(ref_index)
  path(amplicon)
  path(target)

  output:
  path("*.pcr_metrics.csv")

  """
  java -jar /picard.jar CollectTargetedPcrMetrics \
       I=$bam \
   O=${pair_id}_pcr_metrics.txt \
       R=$ref \
   AMPLICON_INTERVALS=$amplicon \
       TARGET_INTERVALS=$target
       
  sed -n -e 7p -e 8p ${pair_id}_pcr_metrics.txt |  sed 's/\t/,/g'   > ${pair_id}.pcr_metrics.csv
  
  

  """


}


workflow {

get_pcr_metric(bam_chl, ref_chl,ref_index_chl, amplicon_chl, target_chl)
}

Answer 1

The issue here is that the fromPath factory method produces a queue channel, but you don't want all of your input channels to be queue channels. Most of the time, what you want is one queue channel and one or more value channels when you require multiple input channels.

You could use the value factory method to create the value channels, but a simpler way is to just pass in some file objects. The latter works because a value channel is implicitly created by a process when it is invoked with a simple value¹. For example:

params.ref_fasta = "./sequences/hg38.fa"
params.amplicon_intervals = "./sequences/tp53_amplicon_intervals.hg38.interval_list"
params.target_intervals = "./sequences/TP53.hg38.interval_list"

params.bam_files = "./bam_files/*.md.bam"
params.outdir = './results'

process get_pcr_metric {

    publishDir path: "${params.outdir}/pcr_metric", mode: 'copy'

    input:
    tuple val(pair_id), path(bam_file)
    path ref_fasta
    path amplicon_intervals
    path target_intervals

    output:
    path "${pair_id}_pcr_metrics.txt"

    """
    picard CollectTargetedPcrMetrics \\
         I="${bam_file}" \\
         O="${pair_id}_pcr_metrics.txt" \\
         R="${ref_fasta}" \\
         AMPLICON_INTERVALS="${amplicon_intervals}" \\
         TARGET_INTERVALS="${target_intervals}"
    """
}

workflow {

    ref_fasta = file( params.ref_fasta )
    amplicon_intervals = file( params.amplicon_intervals )
    target_intervals = file( params.target_intervals )

    Channel
        .fromPath( params.bam_files )
        .map { tuple( it.getBaseName(2), it ) }
        .set { bam_files }

    get_pcr_metric( bam_files, ref_fasta, amplicon_intervals, target_intervals )
}

Answer 2

The code can be made to work on all samples by changing the interface of the process and using the combine operator:

ref_chl = Channel
      .fromPath("sequences/hg38.fa")
ref_index_chl = Channel
      .fromPath("sequences/hg38.fa.fai")
amplicon_chl = Channel
      .fromPath("sequences/tp53_amplicon.hg38.interval_list")
target_chl = Channel
      .fromPath("sequences/TP53.hg38.interval_list")

bam_chl = Channel
     .fromPath("bam_files/*.md.bam")
     .map{ item -> tuple(item.baseName.substring(0, 18), item) }
     .combine(ref_chl)
     .combine(ref_index_chl)
     .combine(amplicon_chl)
     .combine(target_chl)

process get_pcr_metric {

  publishDir path: 'pcr_metric', pattern: '*.csv'
  
  input:
  tuple val(pair_id), path(bam), path(ref), path(ref_index), path(amplicon), path(target)

  output:
  path("*.pcr_metrics.csv")

  """
  java -jar /picard.jar CollectTargetedPcrMetrics \
       I=$bam \
   O=${pair_id}_pcr_metrics.txt \
       R=$ref \
   AMPLICON_INTERVALS=$amplicon \
       TARGET_INTERVALS=$target
       
  sed -n -e 7p -e 8p ${pair_id}_pcr_metrics.txt |  sed 's/\t/,/g'   > ${pair_id}.pcr_metrics.csv
  """
}

workflow {
  get_pcr_metric(bam_chl)
}

The combine operator as the name implies generates all combinations. The different file objects are then in one tuple.

Since only one Channel is passed to the process, it's now also possible to rewrite the workflow code as follows:

workflow {

  refs = [
    "sequences/hg38.fa",
    "sequences/hg38.fa.fai",
    "sequences/tp53_amplicon.hg38.interval_list",
    "sequences/TP53.hg38.interval_list"
  ]

  ref_chl = Channel.from(refs)
   | map{ pointer -> file(pointer) }
   | collect

  Channel.fromPath("bam_files/*.md.bam")
   | map{ item -> tuple(item.baseName.substring(0, 18), item) }
   | combine( ref_chl )
   | get_pcr_metric

}