Cloud object storage support

[mwiewior]

Hello - thanks a lot for this great set of crates!
I'm trying to figure out what would be the best way of using noodles to access files stored on various cloud object storages. I'm looking for an analogy of fsspec in Python I found this crate: object_store
I'm wondering if everyone tried to use it or have some examples of other integrations. I think it would be great to have similar kind of unified layer of accessing files stored in the cloud.

Thanks,
Marek

[brainstorm]

For AWS we wrote this with aws-sdk-rust, with a modular storage trait/layer:

https://github.com/umccr/htsget-rs

But for a more general storage access crate I might try OpenDAL?:

https://github.com/apache/opendal

Hope this helps!

[mwiewior]

@brainstorm thanks!
I actually tried both opendal and object_store and failed to get them working in a "streaming" mode. Do you maybe have any working example with a stream reader for VCF or BAM files ?

[brainstorm]

@brainstorm thanks! I actually tried both opendal and object_store and failed to get them working in a "streaming" mode. Do you maybe have any working example with a stream reader for VCF or BAM files ?

I haven't one at hand with neither of those crates (didn't know the object_store TRAIT crate btw, cool one)... Instead, perhaps we can assist you if you move this thread issue to a discussion and share your attempts/code there?

[zaeleus]

noodles provides readers where the underlying source is generic over either synchronous read traits (provided by std::io) or asynchronous read traits (provided by tokio::io).

In the case of, e.g., OpenDAL, it only supports futures::io traits. However, we can use its bytes::Byte stream via an adapter, i.e., tokio_util::io::StreamReader, and compose it with an async format reader in noodles.

Here is a full example of streaming a BAM file from a public GCS bucket using OpenDAL:

main.rs

// cargo add noodles@0.91.0 --features async,bam,sam
// cargo add opendal@0.51.2 --features services-gcs
// cargo add tokio@1.43.0 --features io-std,macros,rt-multi-thread
// cargo add tokio-util@0.7.13 --features io
// cargo add futures@0.3.31 --no-default-features --features std

use futures::TryStreamExt;
use noodles::{bam, sam};
use opendal::{services::Gcs, Operator};
use tokio::io::{self, AsyncWriteExt};
use tokio_util::io::StreamReader;

const BUCKET: &str = "genomics-public-data";
const NAME: &str = "1000-genomes/bam/HG00096.mapped.ILLUMINA.bwa.GBR.low_coverage.20120522.bam";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let builder = Gcs::default()
        .bucket(BUCKET)
        .disable_vm_metadata()
        .allow_anonymous();

    let operator = Operator::new(builder)?.finish();

    let stream = operator.reader(NAME).await?.into_bytes_stream(..).await?;
    let inner = StreamReader::new(stream);
    let mut reader = bam::r#async::io::Reader::new(inner);

    let header = reader.read_header().await?;

    let stdout = io::stdout();
    let mut writer = sam::r#async::io::Writer::new(stdout);

    writer.write_header(&header).await?;

    let mut records = reader.records();

    while let Some(record) = records.try_next().await? {
        writer.write_alignment_record(&header, &record).await?;
    }

    writer.get_mut().shutdown().await?;

    Ok(())
}

[mwiewior]

@zaeleus thank you a lot !
Your example for BAM reading works like a charm, however if I modify just a little to read VCF file like this:

const BUCKET: &str = "gcp-public-data--gnomad";
const NAME: &str = "release/4.1/vcf/exomes/gnomad.exomes.v4.1.sites.chr21.vcf.bgz";

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>>{


    let builder = Gcs::default()
        .bucket(BUCKET)
        .disable_vm_metadata()
        .allow_anonymous();

    let operator = Operator::new(builder)?.finish();

    let stream = operator.reader(NAME).await?.into_bytes_stream(..).await?;
    let inner = StreamReader::new(stream);
    let mut reader = vcf::r#async::io::Reader::new(inner);

    let header = reader.read_header().await?;

    println!("{:?}", header);

    Ok(())
}

I get the error:

Error: Custom { kind: InvalidData, error: Empty }

and this was actually the problem I was facing before with the object_store crate and a code like this:

object_store = {version = "0.11.2", features = ["gcp"] }

const BUCKET: &str = "gcp-public-data--gnomad";
const NAME: &str = "release/4.1/vcf/exomes/gnomad.exomes.v4.1.sites.chr21.vcf.bgz";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>>{

    let gcs = GoogleCloudStorageBuilder::default().with_bucket_name(BUCKET).build()?;
    let path = Path::from(NAME);
    let result = gcs.get(&path).await?;
    let stream = result.into_stream();
    let inner = StreamReader::new(stream);
    let mut reader = vcf::r#async::io::Reader::new(inner);

    let header = reader.read_header().await?;

    println!("{:?}", header);

    Ok(())
}

Error: Custom { kind: InvalidData, error: Empty }

Do you have any idea what might be wrong here?

[zaeleus]

Your example input is bgzip-compressed (*.bgz) and requires a decoder (noodles::bgzf::r#async::Reader), e.g.,

let stream = operator.reader(NAME).await?.into_bytes_stream(..).await?;
let inner = bgzf::r#async::Reader::new(StreamReader::new(stream));
let mut reader = vcf::r#async::io::Reader::new(inner);

let header = reader.read_header().await?;

let stdout = io::stdout();
let mut writer = vcf::r#async::io::Writer::new(stdout);

writer.write_header(&header).await?;

[mwiewior]

yes! this was it ! thank you @zaeleus and @brainstorm !