Nextflow mixing the ordering of CSV-file rows when using subworkflows

[mribeirodantas]

It is not really clear to me why, but Nextflow is not respecting the order of rows in the example below. I could try to guess what's going on if we had maxForks > 1, but with it set to 1, I have no clue.

csvinputs = Channel
  .of('a,b,c,d,e,f\n10,10,10,10,10,10\n20,20,20,20,20,20\n30,30,30,30,30,30')
  .splitCsv(header: true)

a = csvinputs.map { row -> row.a }
b = csvinputs.map { row -> row.b }

process PROCESS1 {
    maxForks 1
    input:
        val a

    output:
        val "${a}", emit: a1

    script:
    """
    #!/usr/bin/env python
    """
}

process PROCESS2 {
    maxForks 1
    input:
        val a1
        val a2

    output:
        val "${a1}", emit: b1

    script:
    """
    #!/usr/bin/env python

    assert ${a1} == ${a2}
    """
}

workflow WORKFLOW1 {
    take:
        a
        b
    main:
        PROCESS1(a)
        PROCESS2(PROCESS1.out.a1, b)
}

workflow {
    WORKFLOW1(a, b)
}

Sometimes, this Nextflow pipeline will work just fine. However, in some other times, it will throw:

➜  ~ nextflow run ex3.nf
N E X T F L O W  ~  version 22.10.3
Launching `ex3.nf` [boring_fourier] DSL2 - revision: 62a797b8a4
executor >  local (5)
executor >  local (5)
[2c/8e9d55] process > WORKFLOW1:PROCESS1 (1) [100%] 3 of 3 ✔
[7c/cfc871] process > WORKFLOW1:PROCESS2 (2) [ 50%] 1 of 2, failed: 1
Error executing process > 'WORKFLOW1:PROCESS2 (1)'

Caused by:
  Process `WORKFLOW1:PROCESS2 (1)` terminated with an error exit status (1)

Command executed:

  #!/usr/bin/env python

  assert 20 == 10

Command exit status:
  1

Command output:
  (empty)

Command error:
  Traceback (most recent call last):
    File ".command.sh", line 3, in <module>
      assert 20 == 10
  AssertionError

Work dir:
  /Users/mribeirodantas/work/d0/52adb1e399d0f2b506983f9afd4f28

Tip: you can try to figure out what's wrong by changing to the process work dir and showing the script file named `.command.sh`

[pditommaso]

I think the maxFork should not be used to force the sequential execution of tasks. nextflow is inherently parallel, therefore it's not guaranteed that different processes will be executed in the expected order (without mention the lack of performance).

A better solution consists in keeping the channel order using the join operator using the task.index value as the joining column. I've rewritten the example for your convenience

csvinputs = Channel
  .of('a,b,c,d,e,f\n10,10,10,10,10,10\n20,20,20,20,20,20\n30,30,30,30,30,30')
  .splitCsv(header: true)

def index=0
a = csvinputs.map { row -> row.a }
b = csvinputs.map { row -> [++index, row.b] }

process PROCESS1 {
    input:
        val(a)

    output:
        tuple val(task.index), val(a), emit: a1

    script:
    """
    #!/usr/bin/env python
    """
}

process PROCESS2 {
    input:
        tuple val(tid), val(a1), val(a2)

    output:
        val "${a1}", emit: b1

    script:
    """
    #!/usr/bin/env python

    assert ${a1} == ${a2}
    """
}

workflow WORKFLOW1 {
    take:
        a
        b
    main:
        PROCESS1(a)
        PROCESS2(PROCESS1.out.a1.join(b))
}

workflow {
    WORKFLOW1(a, b)
}

Note, also that "${a}" is a useless (and potentially dangerous) practice. Use a instead

[mribeirodantas]

Thanks, Paolo!