let rec fastq_sample_step ~compiler (fs : fastq_sample pipeline) =     let {work_dir; machine ; _ } = compiler in       match fs with       | Bam_to_fastq (how, what) ->         let bam = compile_aligner_step ~compiler what in         let sample_type =           match how with `Single -> `Single_end | `Paired -> `Paired_end in         let fastq_pair =           let output_prefix = work_dir // to_file_prefix ?read:None what in           Picard.bam_to_fastq ~run_with:machine ~sample_type             ~output_prefix bam         in         fastq_pair       | Paired_end_sample (info, l1, l2) ->         let r1 = fastq_step ~compiler ~read:(`R1 info.fragment_id) l1 in         let r2 = fastq_step ~compiler ~read:(`R2 info.fragment_id) l2 in         let open KEDSL in         workflow_node (fastq_reads ~host:Machine.(as_host machine)                          ~name:info.sample_name                          r1#product#path (Some r2#product#path))           ~name:(sprintf "Pairing sample %s (%s and %s)"                    info.sample_name                    (Filename.basename r1#product#path)                    (Filename.basename r2#product#path))           ~edges:[ depends_on r1; depends_on r2 ]       | Single_end_sample (info, single) ->         let r1 = fastq_step ~compiler ~read:(`R1 info.fragment_id) single in         let open KEDSL in         workflow_node (fastq_reads ~host:Machine.(as_host machine)                          ~name:info.sample_name                          r1#product#path None)           ~equivalence:`None           ~edges:[ depends_on r1 ]           ~name:(sprintf "single-end %s (%s)"                    info.sample_name                    (Filename.basename r1#product#path))       | With_metadata (metadata_spec, p) ->         fastq_sample_step ~compiler p |> apply_with_metadata ~metadata_spec   and compile_aligner_step ~compiler (pipeline : bam pipeline) =     let {reference_build; work_dir; machine; _} = compiler in     let result_prefix = work_dir // to_file_prefix pipeline in     dbg "Result_Prefix: %S" result_prefix;     let rec parallelize_alignment ~make_aligner sample =       match sample with       | Paired_end_sample (info,                            Gunzip_concat r1_list,                            Gunzip_concat r2_list) ->         let exploded =           let count = ref 0 in           List.map2 r1_list r2_list             ~f:(fun r1 r2 ->                 match r1, r2 with                 | (Fastq_gz wf1, Fastq_gz wf2) ->                   let new_info =                     incr count;                     {info with                      fragment_id =                        (* fragmenting = creating fragments of previous fragment *)                        sprintf "%s-fragment-%d" info.fragment_id !count} in                   make_aligner (                     Paired_end_sample (new_info,                                        Gunzip_concat [r1],                                        Gunzip_concat [r2]))                 | other -> failwith "compile_aligner_step: not implemented"               ) in         let bams = List.map exploded ~f:(compile_aligner_step ~compiler) in         Samtools.merge_bams ~run_with:machine bams           (result_prefix ^ "-merged.bam")       | With_metadata (metadata_spec, p) ->         parallelize_alignment ~make_aligner p         |> apply_with_metadata ~metadata_spec       | other ->         failwith "parallelize_alignment: Not fully implemented"     in     let catch_parallelize_aligner aligner sample =       let option =         List.find_map compiler.options           (function           | `Parallel_alignment_over_fastq_fragments (al, todo)             when List.mem ~set:al aligner -> Some todo           | _ -> None)       in       match sample with       | Paired_end_sample (name,                            Gunzip_concat r1_list,                            Gunzip_concat r2_list)         when List.length r1_list > 1 ->         begin match option with         | Some _ -> `Caught         | None -> `Not_caught         end       | Paired_end_sample (name,                            Gunzip_concat r1_list,                            Gunzip_concat r2_list)         when List.length r1_list = 1 -> `Not_caught       | other ->         begin match option with         | Some `Silent         | None -> `Not_caught         | Some `Fail_if_not_happening ->           failwithf "Option Parallel_alignment_over_fastq_fragments is set to Fail_if_not_happening and it didn't happen:\n%s"             (to_json other |> Yojson.Basic.pretty_to_string)         end     in     let perform_aligner_parallelization aligner_tag ~make_aligner         ~make_workflow sample =       begin match catch_parallelize_aligner aligner_tag sample  with       | `Caught -> parallelize_alignment ~make_aligner sample       | `Not_caught ->         let fastq = fastq_sample_step ~compiler sample in         make_workflow fastq       end     in     let bam_node =       match pipeline with       | Bam_sample (name, bam_target) -> bam_target       | Gatk_indel_realigner (configuration, bam)         when has_option compiler ((=) (`Map_reduce `Gatk_indel_realigner)) ->         let input_bam = compile_aligner_step ~compiler bam in         Gatk.indel_realigner_map_reduce ~run_with:machine ~compress:false           ~configuration (KEDSL.Single_bam input_bam)       | Gatk_indel_realigner (configuration, bam) ->         let input_bam = compile_aligner_step ~compiler bam in         Gatk.indel_realigner ~run_with:machine ~compress:false           ~configuration (KEDSL.Single_bam input_bam)       | Gatk_bqsr (configuration, bam) ->         let input_bam = compile_aligner_step ~compiler bam in         let output_bam = result_prefix ^ ".bam" in         Gatk.base_quality_score_recalibrator           ~configuration ~run_with:machine ~input_bam ~output_bam       | Picard_mark_duplicates (settings, bam) ->         let input_bam = compile_aligner_step ~compiler bam in         let output_bam = result_prefix ^ ".bam" in         Picard.mark_duplicates ~settings           ~run_with:machine ~input_bam output_bam       | Bwa_mem (bwa_mem_config, fq_sample) ->         perform_aligner_parallelization           `Bwa_mem ~make_aligner:(fun pe -> Bwa_mem (bwa_mem_config, pe))           fq_sample           ~make_workflow:(fun fastq ->               Bwa.mem_align_to_sam ~reference_build                 ~configuration:bwa_mem_config                 ~fastq ~result_prefix ~run_with:machine ()               |> Samtools.sam_to_bam ~reference_build ~run_with:machine)       | Bwa (bwa_config, what) ->         perform_aligner_parallelization           `Bwa ~make_aligner:(fun pe -> Bwa (bwa_config, pe)) what           ~make_workflow:(fun fastq ->               Bwa.align_to_sam ~reference_build                 ~configuration:bwa_config                 ~fastq ~result_prefix ~run_with:machine ()               |> Samtools.sam_to_bam ~reference_build ~run_with:machine)       | Mosaik (what) ->         perform_aligner_parallelization           `Mosaik ~make_aligner:(fun pe -> Mosaik (pe)) what           ~make_workflow:(fun fastq -> Mosaik.align ~reference_build                 ~fastq ~result_prefix ~run_with:machine ())       | Star (configuration, what) ->         perform_aligner_parallelization           `Star ~make_aligner:(fun pe -> Star (configuration, pe)) what           ~make_workflow:(fun fastq ->               Star.align ~configuration ~reference_build                 ~fastq ~result_prefix ~run_with:machine ())       | Hisat (configuration, what) ->         perform_aligner_parallelization           `Hisat ~make_aligner:(fun pe -> Hisat (configuration, pe)) what           ~make_workflow:(fun fastq ->               Hisat.align ~configuration ~reference_build                 ~fastq ~result_prefix ~run_with:machine ()               |> Samtools.sam_to_bam ~reference_build ~run_with:machine             )       | With_metadata (metadata_spec, p)  ->         compile_aligner_step ~compiler p         |> apply_with_metadata ~metadata_spec     in     compiler.wrap_bam_node pipeline bam_node