WIP: Only stage necessary files

This branch is an attempt to only stage the necessary files.
This would be very efficient on S3, but it's also rather complicated
and involves a lot of hacks. It is also unnecessary on systems
with a shared files system or fusion.

Author: grst

assets/schema_input.json

@@ -13,13 +13,17 @@
                 "errorMessage": "Sample name must be provided and cannot contain spaces",
                 "meta": ["id"]
             },
+            "path": {
+                "type": "string",
+                "format": "path"
+            },
             "checksums": {
                 "type": "string",
                 "format": "file-path",
                 "exists": true,
                 "errorMessage": "FastQ file for reads 1 must be provided, cannot contain spaces and must have extension '.fq.gz' or '.fastq.gz'"
             }
         },
-        "required": ["id", "checksums"]
+        "required": ["id", "path", "checksums"]
     }
 }

modules/local/sha256sum/main.nf

@@ -9,13 +9,15 @@ process SHA256SUM_CHECK {
         'nf-core/ubuntu:20.04' }"
 
     input:
-    tuple val(meta), path(checksum_file)
+    tuple val(meta), path(checksum_file), path("rename.sh"), path("files/??????"),
 
     output:
     tuple val(meta), path(report)
 
     script:
     """
-    sha256sum -c ${checksum_file} > ${meta.id}.report.txt
+    bash rename.sh
+
+    cd work && sha256sum -c ${checksum_file} > ../${meta.id}.report.txt
     """
 }

subworkflows/local/checksum_verify.nf

@@ -16,6 +16,25 @@ include { methodsDescriptionText } from '../local/utils_nfcore_datasync_pipeline
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 */
 
+def splitChecksumFile(f, batchsize) {
+    lines = []
+    f.eachLine { line, index ->
+        def parts = line.split(/\s+/)
+        lines.add([parts[0], parts[1], String.format('%06d', (index % batchsize) + 1)])
+    }
+    return lines.collate(batchsize)
+}
+
+def makeRenameScript(batch) {
+    script = []
+    script.add("#!/bin/bash -euo pipefail")
+    script.add("mkdir -p work")
+    batch.each { checksum, filename, index ->
+        script.add("mkdir -p work/\$(dirname '${filename}') && mv files/${index} 'work/${filename}'")
+    }
+    return script.join("\n")
+}
+
 workflow CHECKSUM_VERIFY {
 
     take:
@@ -26,6 +45,28 @@ workflow CHECKSUM_VERIFY {
     ch_versions = Channel.empty()
     ch_multiqc_files = Channel.empty()
 
+    ch_batches = ch_samplesheet.map{ meta, path, checksum_file ->
+        splitChecksumFile(checksum_file, params.chunksize).withIndex().collect {
+            chunk, index -> [meta, chunk]
+        }
+    }.flatMap { meta, chunk -> tuple(meta, chunk)}
+    // ch_batches.view()
+    ch_scripts = ch_batches.map{ meta, chunk -> [meta, makeRenameScript(chunk)] }
+    // ch_scripts.view()
+    ch_files = ch_batches.join(ch_samplesheet).map{ meta, chunk, path, checksum_file ->
+        [meta, chunk.collect{ checksum, filename, idx -> file("${path}/${filename}")}]
+    }
+    ch_files.view()
+    // ch_files = ch_batches.map{
+    //     meta, path, chunk -> chunk.each{
+    //         checksum, filename, numeric_id ->
+    //             files = []
+    //             files.add(file("${path}/${filename}", checkIfExists:true))
+    //     }
+    // }
+    // ch_files.view()
+    // SHA256SUM_CHECK([[:], ["foo/test.txt", "foo/bar.txt"], [file("foo/test.txt"), file("foo/bar.txt")], []])
+
     emit:
     versions      = ch_versions                 // channel: [ path(versions.yml) ]
     multiqc_files = ch_multiqc_files

subworkflows/local/utils_nfcore_datasync_pipeline/main.nf

@@ -74,7 +74,6 @@ workflow PIPELINE_INITIALISATION {
 
     Channel
         .fromList(samplesheetToList(params.input, "${projectDir}/assets/schema_input.json"))
-        .groupTuple()
         .map { samplesheet ->
             validateInputSamplesheet(samplesheet)
         }