How do I aggregate tuples properly?

[rongcuid]

I don't know how to properly ask this question, but let me show an example.

process experiment {
  output:
     tuple path("result.json"), val(metadata)
}
process aggregate {
  input:
    val allExperiments
  output:
    path "aggregate.out"
   """
   for //???
     do something
   end
   """
}

workflow {
  experiment | collect | aggregate
}

Here, there are multiple experiments, each generating its own output file and some metadata. In the aggregate process, I want to process them into a single file (I am using DuckDB here). However, I cannot find a way to make this happen. I thought of multiple possibilities:

1. For loop in Groovy directly. However, script block takes a single string output, so this doesn't work
2. For loop in script. I don't know how to do it because input is a tuple
3. Use multiple process or each. However, I don't know how to make all of them output the same file
4. Write a seprate script and pass in input file as parameter. However, I have too many files and it will exceed maximum command line length

I am certain that this must be doable, but I just don't know how to do so. What's the proper way to perform this operation?

Preferrably I'd like to not change the experiment process, because it has already been run and I don't want to rerun it.

[bentsherman]

What I would do is to transpose the inputs before aggregate so that you get a single tuple with list of metadata and list of paths. The input to aggregate would then be something like tuple path(results_list), val(metadata_list). Then you can iterate through them with a loop in the bash script for aggregate.

[rongcuid]

Is this what you are referring to? It makes the first error:

exps = Channel.of(1,2,3,4,5)

process experiment {
    input:
        val id
    output:
        tuple path("main.txt"), val(id)
    """
    echo $id > main.txt
    echo A > meta.txt
    """
}

process analyze {
    input:
        tuple path(outs), val(meta_list)
    output:
        path "analyze.txt"
    """
    cat main.txt* > analyze.txt
    """
}

workflow {
    exps | experiment | collect | map {it.transpose()} | analyze
}

Also, the cat is only a placeholder. The actual script requires a for loop, which is quite hard to do in bash because I need to somehow pass them from Nextflow to the script.

[rongcuid]

Ok, that particular error was caused by not using collect(flat:false). However, I encounter a new error:

Caused by:
  Process `analyze` input file name collision -- There are multiple input files for each of the following file names: main.txt

[bentsherman]

I think you need to change how the files are staged into analyze:

process analyze {
    input:
        tuple path('*.main.txt'), val(meta_list)
    output:
        path "analyze.txt"
    """
    cat *.main.txt > analyze.txt
    """
}

This way I believe they will be staged as 1.main.txt, 2.main.txt, etc

[rongcuid]

That works. So the final thing (I hope) is that I need a way to loop over each pair of file/metadata and use it in the script. Do I have to use the bash for loop for that?

[bentsherman]

That's what I recommend