Implement common schemas

[jagedn]

Following #11 one of the most "difficult" part of the plugin is create and maintain Java record for schemas

Maybe can be interesting the plugin can offer some common schemas out-the-box as for example FastQRecord (and also another more generics records as StringMap)

for example, a pipeline to convert from raw Fasta text to parquet:

include { toParquet; toFastq } from 'plugin/nf-parquet'

workflow {
    channel.fromPath('data/HI.4549.004.index_10.ANN0830_R2.fastq')
        .splitFastq( record: true )
        .map{ record ->
            toFastq(record)
        }
        .toParquet('data/HI.4549.004.index_10.ANN0830_R2.parquet', ['schema':'fastq'])
        .view{ record -> record }

}

here, the pipeline is using splitFastq to read a fastq file, convert the Map version of Nextflow into an internal Java Record , and write the Java records to a parquet file using schema instead of record parameter

schema can be one of:

• stringMap
• fastq
• fasta
• ....

Custom/propietary formats will be out the scope of this issue and will required the current approach

@Hoohm , what do you think? can cover your use case ?

[Hoohm]

That's pretty interesting, we actually currently do some extraction from fastq to parquet as want do some string manipulation for barcode detection/correction. We're currently using polars_bio for that.

The main issue we will be facing is that our schema used for extraction from fastq is dynamic based on the chemistry choice. ChemistryA might have 4 string columns, chemistryB might have 6 with different names. So we'll need to implement a mechanism to somehow inject those schemas at runtime or before.

The source of schema is currently in a json file atm, linked to the chemistry_id. We could add a string that could be interpreted although this seems pretty brittle.

Another possibility is to generate a record file and load that somehow at runtime.

Template schemas could be fetched by simple reserved string names and custom ones could be pointed to a custom record file. The record file could be generated before runtime (easier) or at runtime and somehow referenced there.

Just to go back to the original interest for chunking parquet files. Most of our code relies on polars lazyFrames but some just needs to be in DataFrames. We're tying to reach some specs in terms of memory usage and for those specific parts, chunking -> process -> merging would be the path forward to not hint memory limits on huge files, hence the interest to do that via nf-parquet. We could also just do it via processes but nf-parquet should make it faster and more elegant.

[jagedn]

I need to think how to create a java record at runtime, not sure how to do it.

About chunks I've playing with this (I hope interesting) small example:

Say we have 3 Java Records, CustomRecord and TopRecord and SimpleRecord.
We want to read a big parquet file and create 2 output parquet files: top and loosers

workflow{

    channel.fromPath("data/customs.parquet")
           .splitParquet(by:params.size, file:true, record:CustomRecord) //origin
           .set { ch_chunk }

    ch_chunk.splitParquet(record:CustomRecord)
          | processCustomRecord
          | branch{
                top: it.id < 20
                loosers: it.id > 19
            }
          | set{ ch_write }

    ch_write.top
          .map{ new TopRecord(it.id, it.name, it.value, it.samples) }
          .toParquet( params.top, [ record:TopRecord] )

    ch_write.loosers
          .map{ new SingleRecord(it.id, it.name) }
          .toParquet( params.loosers, [record:SingleRecord] )
}


process processCustomRecord {  // process example
    input:
    val 'custom'

    output:
    val custom

    exec:
    println "do whatever with the record $custom"

}