Merge pull request #998 from FriederikeHanssen/issue_952

Issue 952

Author: maxulysse

CHANGELOG.md

@@ -50,6 +50,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - [#982](https://github.com/nf-core/sarek/pull/982) - Remove usage of exit statements, using `Nextflow.error` instead
 - [#985](https://github.com/nf-core/sarek/pull/985) - Cache correctly identifies when it needs to be updated
 - [#988](https://github.com/nf-core/sarek/pull/988) - Updated ascat module to fix seed for reproducibility
+- [#998](https://github.com/nf-core/sarek/pull/998) - Remove parallelization within a sample for `Manta`
 
 ### Deprecated
 

conf/modules/manta.config

@@ -16,42 +16,14 @@
 process {
 
     withName: 'MANTA.*' {
-        ext.prefix       = { meta.num_intervals <= 1 ? "${meta.id}.manta" : "${meta.id}.manta.${target_bed.simpleName}" }
         ext.args         = { params.wes ? "--exome" : "" }
+        ext.prefix       = { "${meta.id}.manta" }
         ext.when         = { params.tools && params.tools.split(',').contains('manta') }
         publishDir       = [
             mode: params.publish_dir_mode,
-            path: { "${params.outdir}/variant_calling/" },
-            pattern: "*{diploid_sv,tumor_sv,somatic_sv}.{vcf.gz,vcf.gz.tbi}",
-            saveAs: { meta.num_intervals > 1 ? null : "manta/${meta.id}/${it}" }
-        ]
-    }
-
-    withName: 'MERGE_MANTA.*' {
-        publishDir       = [
-            mode: params.publish_dir_mode,
-            path: { "${params.outdir}/variant_calling/manta/${meta.id}/" },
+            path: { "${params.outdir}/variant_calling/manta/${meta.id}" },
             pattern: "*{diploid_sv,tumor_sv,somatic_sv}.{vcf.gz,vcf.gz.tbi}"
         ]
     }
 
-    withName: 'MERGE_MANTA_DIPLOID' {
-        ext.prefix       = {"${meta.id}.manta.diploid_sv"}
-    }
-
-    withName: 'MERGE_MANTA_SMALL_INDELS' {
-        ext.prefix       = {"${meta.id}.manta.candidate_small_indels"}
-    }
-
-    withName: 'MERGE_MANTA_SV' {
-        ext.prefix       = {"${meta.id}.manta.candidate_sv"}
-    }
-
-    withName: 'MERGE_MANTA_TUMOR' {
-        ext.prefix       = {"${meta.id}.manta.tumor_sv"}
-    }
-
-    withName: 'MERGE_MANTA_SOMATIC' {
-        ext.prefix       = {"${meta.id}.manta.somatic_sv"}
-    }
 }

modules.json

@@ -295,12 +295,12 @@
                     },
                     "manta/somatic": {
                         "branch": "master",
-                        "git_sha": "c8e35eb2055c099720a75538d1b8adb3fb5a464c",
+                        "git_sha": "f68f379dcdba4e8c132c6daa6cd8794527b81a04",
                         "installed_by": ["modules"]
                     },
                     "manta/tumoronly": {
                         "branch": "master",
-                        "git_sha": "c8e35eb2055c099720a75538d1b8adb3fb5a464c",
+                        "git_sha": "f68f379dcdba4e8c132c6daa6cd8794527b81a04",
                         "installed_by": ["modules"]
                     },
                     "mosdepth": {

modules/nf-core/manta/somatic/main.nf

@@ -26,6 +26,7 @@ workflow BAM_VARIANT_CALLING_GERMLINE_ALL {
     fasta_fai                         // channel: [mandatory] fasta_fai
     intervals                         // channel: [mandatory] [ intervals, num_intervals ] or [ [], 0 ] if no intervals
     intervals_bed_combined            // channel: [mandatory] intervals/target regions in one file unzipped
+    intervals_bed_gz_tbi_combined     // channel: [mandatory] intervals/target regions in one file zipped
     intervals_bed_combined_haplotypec // channel: [mandatory] intervals/target regions in one file unzipped, no_intervals.bed if no_intervals
     intervals_bed_gz_tbi              // channel: [mandatory] [ interval.bed.gz, interval.bed.gz.tbi, num_intervals ] or [ [], [], 0 ] if no intervals
     known_indels_vqsr
@@ -159,7 +160,7 @@ workflow BAM_VARIANT_CALLING_GERMLINE_ALL {
             dict.map{ it -> [ [ id:'dict' ], it ] },
             fasta,
             fasta_fai,
-            intervals_bed_gz_tbi
+            intervals_bed_gz_tbi_combined
         )
 
         vcf_manta = BAM_VARIANT_CALLING_GERMLINE_MANTA.out.vcf

modules/nf-core/manta/tumoronly/main.nf

@@ -1,6 +1,3 @@
-include { GATK4_MERGEVCFS as MERGE_MANTA_DIPLOID      } from '../../../modules/nf-core/gatk4/mergevcfs/main'
-include { GATK4_MERGEVCFS as MERGE_MANTA_SMALL_INDELS } from '../../../modules/nf-core/gatk4/mergevcfs/main'
-include { GATK4_MERGEVCFS as MERGE_MANTA_SV           } from '../../../modules/nf-core/gatk4/mergevcfs/main'
 include { MANTA_GERMLINE                              } from '../../../modules/nf-core/manta/germline/main'
 
 // Seems to be the consensus on upstream modules implementation too
@@ -10,57 +7,29 @@ workflow BAM_VARIANT_CALLING_GERMLINE_MANTA {
     dict          // channel: [optional]  [ meta, dict ]
     fasta         // channel: [mandatory] [ fasta ]
     fasta_fai     // channel: [mandatory] [ fasta_fai ]
-    intervals     // channel: [mandatory] [ interval.bed.gz, interval.bed.gz.tbi, num_intervals ] or [ [], [], 0 ] if no intervals
+    intervals     // channel: [mandatory] [ interval.bed.gz, interval.bed.gz.tbi] or [ [], []] if no intervals; intervals file contains all intervals
 
     main:
     versions = Channel.empty()
 
-    // Combine cram and intervals for spread and gather strategy
-    cram_intervals = cram.combine(intervals)
-        // Move num_intervals to meta map
-        .map{ meta, cram, crai, intervals, intervals_index, num_intervals -> [ meta + [ num_intervals:num_intervals ], cram, crai, intervals, intervals_index ] }
+    // Combine cram and intervals, account for 0 intervals
+    cram_intervals = cram.combine(intervals).map{ it ->
+        bed_gz = it.size() > 3 ? it[3] : []
+        bed_tbi = it.size() > 3 ? it[4] : []
 
-    MANTA_GERMLINE(cram_intervals, fasta, fasta_fai)
-
-    // Figuring out if there is one or more vcf(s) from the same sample
-    small_indels_vcf = MANTA_GERMLINE.out.candidate_small_indels_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
-    }
-
-    // Figuring out if there is one or more vcf(s) from the same sample
-    sv_vcf = MANTA_GERMLINE.out.candidate_sv_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
+        [it[0], it[1], it[2], bed_gz, bed_tbi]
     }
 
-    // Figuring out if there is one or more vcf(s) from the same sample
-    diploid_sv_vcf = MANTA_GERMLINE.out.diploid_sv_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
-    }
-
-    // Only when using intervals
-    diploid_sv_vcf_to_merge = diploid_sv_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
-    small_indels_vcf_to_merge = small_indels_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
-    sv_vcf_to_merge = sv_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
+    MANTA_GERMLINE(cram_intervals, fasta, fasta_fai)
 
-    MERGE_MANTA_DIPLOID(diploid_sv_vcf_to_merge, dict)
-    MERGE_MANTA_SMALL_INDELS(small_indels_vcf_to_merge, dict)
-    MERGE_MANTA_SV(sv_vcf_to_merge, dict)
+    small_indels_vcf = MANTA_GERMLINE.out.candidate_small_indels_vcf
+    sv_vcf = MANTA_GERMLINE.out.candidate_sv_vcf
+    diploid_sv_vcf = MANTA_GERMLINE.out.diploid_sv_vcf
 
-    // Mix intervals and no_intervals channels together
     // Only diploid SV should get annotated
-    vcf = Channel.empty().mix(MERGE_MANTA_DIPLOID.out.vcf, diploid_sv_vcf.no_intervals)
-        // add variantcaller to meta map and remove no longer necessary field: num_intervals
-        .map{ meta, vcf -> [ meta - meta.subMap('num_intervals') + [ variantcaller:'manta' ], vcf ] }
+    // add variantcaller to meta map
+    vcf = diploid_sv_vcf.map{ meta, vcf -> [ meta + [ variantcaller:'manta' ], vcf ] }
 
-    versions = versions.mix(MERGE_MANTA_DIPLOID.out.versions)
-    versions = versions.mix(MERGE_MANTA_SMALL_INDELS.out.versions)
-    versions = versions.mix(MERGE_MANTA_SV.out.versions)
     versions = versions.mix(MANTA_GERMLINE.out.versions)
 
     emit:

subworkflows/local/bam_variant_calling_germline_all/main.nf

@@ -31,6 +31,7 @@ workflow BAM_VARIANT_CALLING_SOMATIC_ALL {
     intervals                     // channel: [mandatory] [ intervals, num_intervals ] or [ [], 0 ] if no intervals
     intervals_bed_gz_tbi          // channel: [mandatory] intervals/target regions index zipped and indexed
     intervals_bed_combined        // channel: [mandatory] intervals/target regions in one file unzipped
+    intervals_bed_gz_tbi_combined // channel: [mandatory] intervals/target regions in one file zipped
     mappability
     msisensorpro_scan             // channel: [optional]  msisensorpro_scan
     panel_of_normals              // channel: [optional]  panel_of_normals
@@ -147,7 +148,7 @@ workflow BAM_VARIANT_CALLING_SOMATIC_ALL {
             dict.map{ it -> [ [ id:'dict' ], it ] },
             fasta,
             fasta_fai,
-            intervals_bed_gz_tbi
+            intervals_bed_gz_tbi_combined
         )
 
         vcf_manta = BAM_VARIANT_CALLING_SOMATIC_MANTA.out.vcf

subworkflows/local/bam_variant_calling_germline_manta/main.nf

@@ -1,7 +1,3 @@
-include { GATK4_MERGEVCFS as MERGE_MANTA_DIPLOID           } from '../../../modules/nf-core/gatk4/mergevcfs/main'
-include { GATK4_MERGEVCFS as MERGE_MANTA_SMALL_INDELS      } from '../../../modules/nf-core/gatk4/mergevcfs/main'
-include { GATK4_MERGEVCFS as MERGE_MANTA_SOMATIC           } from '../../../modules/nf-core/gatk4/mergevcfs/main'
-include { GATK4_MERGEVCFS as MERGE_MANTA_SV                } from '../../../modules/nf-core/gatk4/mergevcfs/main'
 include { MANTA_SOMATIC                                    } from '../../../modules/nf-core/manta/somatic/main'
 
 workflow BAM_VARIANT_CALLING_SOMATIC_MANTA {
@@ -10,86 +6,31 @@ workflow BAM_VARIANT_CALLING_SOMATIC_MANTA {
     dict          // channel: [optional]  [ meta, dict ]
     fasta         // channel: [mandatory] [ fasta ]
     fasta_fai     // channel: [mandatory] [ fasta_fai ]
-    intervals     // channel: [mandatory] [ interval.bed.gz, interval.bed.gz.tbi, num_intervals ] or [ [], [], 0 ] if no intervals
+    intervals     // channel: [mandatory] [ interval.bed.gz, interval.bed.gz.tbi ] or [ [], [] ] if no intervals
 
     main:
     versions = Channel.empty()
 
-    // Combine cram and intervals for spread and gather strategy
-    cram_intervals = cram.combine(intervals)
-        // Move num_intervals to meta map
-        .map{ meta, cram1, crai1, cram2, crai2, intervals, intervals_index, num_intervals -> [ meta + [ num_intervals:num_intervals ], cram1, crai1, cram2, crai2, intervals, intervals_index ] }
+    // Combine cram and intervals, account for 0 intervals
+    cram_intervals = cram.combine(intervals).map{ it ->
+        bed_gz = it.size() > 5 ? it[5] : []
+        bed_tbi = it.size() > 5 ? it[6] : []
 
-    MANTA_SOMATIC(cram_intervals, fasta, fasta_fai)
-
-    // Figuring out if there is one or more vcf(s) from the same sample
-    candidate_small_indels_vcf = MANTA_SOMATIC.out.candidate_small_indels_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
-    }
-
-    // Figuring out if there is one or more vcf(s) from the same sample
-    candidate_small_indels_vcf_tbi = MANTA_SOMATIC.out.candidate_small_indels_vcf_tbi.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
+        [it[0], it[1], it[2], it[3], it[4], bed_gz, bed_tbi]
     }
 
-    // Figuring out if there is one or more vcf(s) from the same sample
-    candidate_sv_vcf = MANTA_SOMATIC.out.candidate_sv_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
-    }
-
-    // Figuring out if there is one or more vcf(s) from the same sample
-    diploid_sv_vcf = MANTA_SOMATIC.out.diploid_sv_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
-    }
-
-    // Figuring out if there is one or more vcf(s) from the same sample
-    somatic_sv_vcf = MANTA_SOMATIC.out.somatic_sv_vcf.branch{
-        // Use meta.num_intervals to asses number of intervals
-        intervals:    it[0].num_intervals > 1
-        no_intervals: it[0].num_intervals <= 1
-    }
-
-    // Only when using intervals
-    candidate_small_indels_vcf_to_merge = candidate_small_indels_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
-    candidate_sv_vcf_to_merge = candidate_sv_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
-    diploid_sv_vcf_to_merge = diploid_sv_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
-    somatic_sv_vcf_to_merge = somatic_sv_vcf.intervals.map{ meta, vcf -> [ groupKey(meta, meta.num_intervals), vcf ]}.groupTuple()
+    MANTA_SOMATIC(cram_intervals, fasta, fasta_fai)
 
-    MERGE_MANTA_SMALL_INDELS(candidate_small_indels_vcf_to_merge, dict)
-    MERGE_MANTA_SV(candidate_sv_vcf_to_merge, dict)
-    MERGE_MANTA_DIPLOID(diploid_sv_vcf_to_merge, dict)
-    MERGE_MANTA_SOMATIC(somatic_sv_vcf_to_merge, dict)
+    candidate_small_indels_vcf = MANTA_SOMATIC.out.candidate_small_indels_vcf
+    candidate_small_indels_vcf_tbi = MANTA_SOMATIC.out.candidate_small_indels_vcf_tbi
+    candidate_sv_vcf = MANTA_SOMATIC.out.candidate_sv_vcf
+    diploid_sv_vcf = MANTA_SOMATIC.out.diploid_sv_vcf
+    somatic_sv_vcf = MANTA_SOMATIC.out.somatic_sv_vcf
 
-    // Mix intervals and no_intervals channels together
     // Only diploid and somatic SV should get annotated
-    vcf = Channel.empty().mix(MERGE_MANTA_DIPLOID.out.vcf, MERGE_MANTA_SOMATIC.out.vcf, diploid_sv_vcf.no_intervals, somatic_sv_vcf.no_intervals)
-        // add variantcaller to meta map and remove no longer necessary field: num_intervals
-        .map{ meta, vcf -> [ meta - meta.subMap('num_intervals') + [ variantcaller:'manta' ], vcf ] }
-
-    // Mix intervals and no_intervals channels together
-    // Only joining reads for StrelkaBP
-    candidate_small_indels_vcf = Channel.empty().mix(MERGE_MANTA_SMALL_INDELS.out.vcf, candidate_small_indels_vcf.no_intervals)
-        // remove no longer necessary field: num_intervals
-        .map{ meta, vcf -> [ meta - meta.subMap('num_intervals'), vcf ] }
-
-    // Mix intervals and no_intervals channels together
-    // Only joining reads for StrelkaBP
-    candidate_small_indels_vcf_tbi = Channel.empty().mix(MERGE_MANTA_SMALL_INDELS.out.tbi, candidate_small_indels_vcf_tbi.no_intervals)
-        // remove no longer necessary field: num_intervals
-        .map{ meta, tbi -> [ meta - meta.subMap('num_intervals'), tbi ] }
+    // add variantcaller to meta map
+    vcf = Channel.empty().mix(diploid_sv_vcf, somatic_sv_vcf).map{ meta, vcf -> [ meta + [ variantcaller:'manta' ], vcf ] }
 
-    versions = versions.mix(MERGE_MANTA_SV.out.versions)
-    versions = versions.mix(MERGE_MANTA_SMALL_INDELS.out.versions)
-    versions = versions.mix(MERGE_MANTA_DIPLOID.out.versions)
-    versions = versions.mix(MERGE_MANTA_SOMATIC.out.versions)
     versions = versions.mix(MANTA_SOMATIC.out.versions)
 
     emit:

subworkflows/local/bam_variant_calling_somatic_all/main.nf

@@ -30,6 +30,7 @@ workflow BAM_VARIANT_CALLING_TUMOR_ONLY_ALL {
     intervals                     // channel: [mandatory] [ intervals, num_intervals ] or [ [], 0 ] if no intervals
     intervals_bed_gz_tbi          // channel: [mandatory] [ interval.bed.gz, interval.bed.gz.tbi, num_intervals ] or [ [], [], 0 ] if no intervals
     intervals_bed_combined        // channel: [mandatory] intervals/target regions in one file unzipped
+    intervals_bed_gz_tbi_combined // channel: [mandatory] intervals/target regions in one file zipped
     mappability
     panel_of_normals              // channel: [optional]  panel_of_normals
     panel_of_normals_tbi          // channel: [optional]  panel_of_normals_tbi
@@ -132,7 +133,8 @@ workflow BAM_VARIANT_CALLING_TUMOR_ONLY_ALL {
             dict.map{ it -> [ [ id:'dict' ], it ] },
             fasta,
             fasta_fai,
-            intervals_bed_gz_tbi
+            intervals_bed_gz_tbi_combined
+
         )
 
         vcf_manta = BAM_VARIANT_CALLING_TUMOR_ONLY_MANTA.out.vcf